sadpandajoe commented on code in PR #43770:
URL: https://github.com/apache/superset/pull/43770#discussion_r4225421693
##########
superset/mcp_service/chart/preview_utils.py:
##########
@@ -1398,6 +2413,120 @@ def fallback_vega_lite_preview(
return None
+def generate_xy_pivot_vega_lite_preview(
+ data: list[dict[str, Any]], form_data: dict[str, Any], *, mark: str
+) -> VegaLitePreview | None:
+ """Render flattened timeseries pivot columns without dropping grouped
series.
+
+ Folding escaped field paths resolves literal output keys without splitting
+ category values that contain escaped commas. The legend retains each
+ complete metric/category label.
+ Long-form results continue through the generic renderer.
+ """
+ from superset.mcp_service.chart.chart_helpers import _as_list,
_time_comparison
+ from superset.utils.pandas_postprocessing.utils import (
+ escape_separator,
+ FLAT_COLUMN_SEPARATOR,
+ )
+
+ def vega_field(name: str) -> str:
+ """Escape a literal key so Vega-Lite does not read it as a nested
path."""
+ return "".join("\\" + char if char in ".[]\\" else char for char in
name)
+
+ if not data:
+ return None
+ dimensions = [
+ label
+ for column in _as_list(form_data.get("groupby"))
+ if (label := _form_column_label(column))
+ ]
+ if not dimensions or any(label in data[0] for label in dimensions):
+ return None
+ x_axis = _form_column_label(form_data.get("x_axis")) or "__timestamp"
+ if x_axis not in data[0]:
+ return None
+ metric_labels = [
+ label
+ for metric in _as_list(form_data.get("metrics"))
+ if (label := metric_result_label(metric))
+ ]
+ # Chart-data results unescape the flattened column names, while raw
+ # post-processing output keeps escaped separators; match either spelling.
+ series_labels = list(metric_labels)
+ if len(metric_labels) == 1 and _time_comparison(
+ form_data, _as_list(form_data.get("metrics"))
+ ):
+ # A single compared metric renames each shifted series to its bare
+ # offset, so those series carry the offset rather than the metric.
+ series_labels.extend(
+ str(offset) for offset in _as_list(form_data.get("time_compare"))
+ )
+ prefixes = {
+ spelling
+ for label in series_labels
+ for spelling in (label, escape_separator(label))
+ }
+ fields = [
+ field
+ for field in data[0]
+ if field != x_axis
+ and any(
+ field.startswith(prefix + FLAT_COLUMN_SEPARATOR)
+ or field.startswith(prefix + "__")
+ for prefix in prefixes
+ )
+ ]
+ if not fields and len(metric_labels) == 1 and
form_data.get("truncate_metric"):
+ # A single truncated metric drops its label from the pivoted column
+ # names, so every non-x-axis column is one category series.
+ fields = [field for field in data[0] if field != x_axis]
+ if not fields:
+ return None
+ sample = data[0][x_axis]
+ x_type = (
+ "temporal"
+ if isinstance(sample, str) and any(char in sample for char in "-/: ")
Review Comment:
Adding a space to this character set makes any grouped category label
containing a space (`New York`, `Product A`) a temporal x-axis, which Vega-Lite
parses as an invalid date, so a bar chart by `product` and `region` previews
with no bars while Explore renders it. ISO dates and timestamps already contain
`-` or `:`, so the space only adds false positives. Can the x type come from
the axis/column type instead of scanning the first value?
##########
superset/mcp_service/chart/chart_utils.py:
##########
@@ -1715,6 +1758,853 @@ def map_histogram_config(config:
"HistogramChartConfig") -> Dict[str, Any]:
return form_data
+def _bullet_token_list(values: Sequence[str | int | float]) -> str:
+ """Serialize typed Bullet controls to the frontend's comma-separated
form."""
+ tokens: list[str] = []
+ for value in values:
+ if isinstance(value, float):
+ token = repr(value)
+ # ``100`` parses back to the same binary float as ``100.0`` and
+ # preserves the frontend's established compact integer spelling.
+ if token.endswith(".0") and not (
+ value == 0.0 and math.copysign(1.0, value) < 0
+ ):
+ token = token[:-2]
+ tokens.append(token)
+ else:
+ tokens.append(str(value))
+ return ",".join(tokens)
+
+
+def map_bullet_config(config: BulletChartConfig) -> Dict[str, Any]: # noqa:
C901
+ """Map typed Bullet config to ``Bullet/buildQuery`` and transformProps.
+
+ The frontend buildQuery replaces the generic query fields with exactly one
+ metric and the groupby hierarchy. Presentation controls stay in native
+ snake_case form_data; the chart plugin camelizes them for transformProps.
+ """
+ if (
+ config.dimensions is None
+ and config._inherited_groupby is None
+ and config.order_by
+ ):
+ # An update resolves its saved hierarchy before mapping. Without one,
+ # creation must validate sort targets against an empty hierarchy.
+ BulletChartConfig.model_validate(
+ {**config.model_dump(exclude_unset=True), "dimensions": []}
+ )
+ metric = create_metric_object(config.metric)
+ form_data: Dict[str, Any] = {
+ "viz_type": "bullet",
+ "metric": metric,
+ }
+
+ # Optional semantic/query fields are emitted only when explicitly supplied.
+ # This lets update_chart and update_chart_preview preserve native saved
state,
+ # while an explicit empty value still clears it through the generic merge
path.
+ # The row limit always carries the schema default so a new chart is
bounded;
+ # update merging restores the saved limit when the caller omits it.
+ if "dimensions" in config.model_fields_set:
+ form_data["groupby"] = [dimension.name for dimension in
config.dimensions or []]
+ form_data["row_limit"] = config.row_limit
+ if "time_range" in config.model_fields_set:
+ form_data["time_range"] = config.time_range
+
+ if config.order_by:
+ dimensions = config.order_dimensions
+ orderby: list[list[Any]] = []
+ for order in config.order_by:
+ role, index = resolve_bullet_order_target(
+ order.column, dimensions, config.metric
+ )
+ if role == "metric":
+ order_target: Any = metric
+ else:
+ if index is None: # Defensive: resolver pairs dimensions with
indexes.
+ raise ValueError("Bullet dimension order target has no
index")
+ dimension = dimensions[index]
+ order_target = (
+ dimension.name if isinstance(dimension, ColumnRef) else
dimension
+ )
+ orderby.append([order_target, order.ascending])
+ form_data["orderby"] = orderby
+ elif "order_by" in config.model_fields_set:
+ form_data["orderby"] = []
+
+ presentation_fields: dict[str, tuple[str, Any]] = {
+ "ranges": ("ranges", _bullet_token_list(config.ranges)),
+ "range_labels": (
+ "range_labels",
+ _bullet_token_list(config.range_labels),
+ ),
+ "markers": ("markers", _bullet_token_list(config.markers)),
+ "marker_labels": (
+ "marker_labels",
+ _bullet_token_list(config.marker_labels),
+ ),
+ "marker_lines": (
+ "marker_lines",
+ _bullet_token_list(config.marker_lines),
+ ),
+ "marker_line_labels": (
+ "marker_line_labels",
+ _bullet_token_list(config.marker_line_labels),
+ ),
+ "y_axis_format": ("y_axis_format", config.y_axis_format),
+ "show_labels": ("show_labels", config.show_labels),
+ "show_legend": ("show_legend", config.show_legend),
+ }
+ for field_name, (form_key, value) in presentation_fields.items():
+ if field_name in config.model_fields_set:
+ form_data[form_key] = value
+
+ _add_adhoc_filters(form_data, config.filters)
+ if config.filters == [] and "filters" in config.model_fields_set:
+ form_data["adhoc_filters"] = []
+ if config.time_range and config.temporal_column:
+ _ensure_temporal_adhoc_filter(form_data, config.temporal_column)
+ for filter_ in form_data.get("adhoc_filters", []):
+ if (
+ isinstance(filter_, dict)
+ and filter_.get("operator") ==
FilterOperator.TEMPORAL_RANGE.value
+ and filter_.get("subject") == config.temporal_column
+ and filter_.get("comparator") == NO_TIME_RANGE
+ ):
+ filter_["comparator"] = config.time_range
+ return form_data
+
+
+def _normalize_native_filter_aliases(form_data: Mapping[str, Any]) ->
Dict[str, Any]:
+ """Fold legacy WHERE, HAVING, and filter predicates into adhoc controls."""
+ from superset.utils.core import form_data_to_adhoc, simple_filter_to_adhoc
+
+ normalized = dict(form_data)
+ legacy_filters = [
+ form_data_to_adhoc(normalized, clause)
+ for clause in ("having", "where")
+ if normalized.get(clause)
+ ]
+ legacy_filters.extend(
+ simple_filter_to_adhoc(filter_, "where")
+ for filter_ in normalized.get("filters") or []
+ if filter_ is not None
+ )
+ if legacy_filters:
+ normalized["adhoc_filters"] = [
+ *legacy_filters,
+ *(normalized.get("adhoc_filters") or []),
+ ]
+ for key in ("where", "having", "filters"):
+ normalized.pop(key, None)
+ return normalized
+
+
+def _normalize_bullet_query_aliases(form_data: Mapping[str, Any]) -> Dict[str,
Any]:
+ """Fold inherited native predicates and ordering into canonical
controls."""
+ from superset.mcp_service.chart.chart_helpers import _parse_orderby
+
+ normalized = _normalize_native_filter_aliases(form_data)
+ if "groupby" in normalized and not isinstance(normalized["groupby"], list):
+ # Bullet/buildQuery and transformProps read ensureIsArray(groupby).
+ normalized["groupby"] = bullet_groupby_list(normalized["groupby"])
+ if "order_by_cols" in normalized:
+ # Native extractQueryFields concatenates both aliases in key order.
+ ordering: list[Any] = []
+ for key, value in normalized.items():
+ if key == "order_by_cols":
+ ordering.extend(_parse_orderby(value))
+ elif key == "orderby":
+ ordering.extend(value or [])
+ normalized["orderby"] = ordering
+ normalized.pop("order_by_cols", None)
+ return normalized
+
+
+def merge_bullet_form_data(
+ existing_form_data: Mapping[str, Any], new_form_data: Dict[str, Any]
+) -> None:
+ """Preserve omitted native Bullet controls across update tool paths.
+
+ Query roles and every UI control have an explicit typed representation.
+ Mappers emit optional fields only when the caller supplied them, so copying
+ the bounded native keys below preserves omitted state while explicit empty,
+ false, null, and zero-like values remain authoritative.
+ """
+ if (
+ existing_form_data.get("viz_type") != "bullet"
+ or new_form_data.get("viz_type") != "bullet"
+ ):
+ return
+ existing_form_data = _normalize_bullet_query_aliases(existing_form_data)
+ preserved_keys = {
+ "groupby",
+ "adhoc_filters",
+ "time_range",
+ "row_limit",
+ "orderby",
+ "ranges",
+ "range_labels",
+ "markers",
+ "marker_labels",
+ "marker_lines",
+ "marker_line_labels",
+ "y_axis_format",
+ "show_labels",
+ "show_legend",
+ "url_params",
+ # Native query context (dashboard/native filter predicates and time
+ # overrides) that buildQueryObject applies on top of the controls.
+ "extra_form_data",
+ "extra_filters",
+ MCP_DASHBOARD_TIME_FILTER_SUBJECT,
+ }
+
+ # Threshold and label arrays are one frontend control pair. If callers
+ # replace the values without replacing their labels, clear the stale labels
+ # instead of accidentally reassigning them by position.
+ dependent_controls = {
+ "ranges": "range_labels",
+ "markers": "marker_labels",
+ "marker_lines": "marker_line_labels",
+ }
+ for values_key, labels_key in dependent_controls.items():
+ if values_key in new_form_data and labels_key not in new_form_data:
+ new_form_data[labels_key] = ""
+
+ preserve_orderby = (
+ "orderby" not in new_form_data and "orderby" in existing_form_data
+ )
+ for key in preserved_keys:
+ if (
+ key == MCP_DASHBOARD_TIME_FILTER_SUBJECT
+ and "adhoc_filters" in new_form_data
+ ):
+ # The marker describes a mapper-generated temporal filter. Do not
+ # retain stale provenance when an explicit filter update removed
it.
+ continue
+ if key in existing_form_data and key not in new_form_data:
+ new_form_data[key] = existing_form_data[key]
+ if preserve_orderby:
+ new_form_data["orderby"] = _orderby_for_final_output_roles(
+ existing_form_data, new_form_data
+ )
+
+
+def _bullet_output_labels(
+ form_data: Mapping[str, Any],
+) -> tuple[set[str], dict[str, Any]]:
+ """Return a Bullet state's dimension output labels and metric outputs."""
+ from superset.mcp_service.chart.chart_helpers import _column_label,
_metric_label
+
+ dimensions = {
+ label
+ for column in bullet_groupby_list(form_data.get("groupby"))
+ if (label := _column_label(column)) is not None
+ }
+ metrics = form_data.get("metrics") or []
+ if not isinstance(metrics, (list, tuple)):
+ metrics = [metrics]
+ metric_outputs = {
+ label: metric
+ for metric in [form_data.get("metric"), *metrics]
+ if (label := _metric_label(metric)) is not None
+ }
+ return dimensions, metric_outputs
+
+
+def bullet_groupby_list(groupby: Any) -> list[Any]:
+ """Normalize a saved Bullet hierarchy like the frontend
``ensureIsArray``."""
+ if groupby is None:
+ return []
+ return list(groupby) if isinstance(groupby, (list, tuple)) else [groupby]
+
+
+def _orderby_for_final_output_roles(
+ existing_form_data: Mapping[str, Any], new_form_data: Mapping[str, Any]
+) -> Any:
+ """Drop sorts on removed output roles and rebind inherited metric
expressions.
+
+ Native ordering may also rank by a saved metric or column that is not a
+ displayed output (``get_sqla_query`` resolves it independently). Those
+ sorters never named a Bullet role, so a role change does not remove them.
+ """
+ from superset.mcp_service.chart.chart_helpers import _column_label,
_metric_label
+
+ saved = existing_form_data.get("orderby")
+ if not isinstance(saved, list):
+ return saved
+ outputs, metric_outputs = _bullet_output_labels(new_form_data)
+ outputs.update(metric_outputs)
+ previous_dimensions, previous_metrics =
_bullet_output_labels(existing_form_data)
+ previous_outputs = previous_dimensions | set(previous_metrics)
+ retained = []
+ for entry in saved:
+ if isinstance(entry, (list, tuple)) and entry:
+ target = entry[0]
+ label = (
+ _metric_label(target)
+ or _column_label(target)
+ or target.get("metric_name")
+ if isinstance(target, Mapping)
+ else target
+ )
+ if (
+ isinstance(label, str)
+ and label not in outputs
+ and label in previous_outputs
+ ):
+ continue
+ if (
+ isinstance(target, Mapping)
+ and isinstance(label, str)
+ and label in metric_outputs
+ ):
+ # Label equality identifies an output role, not expression
+ # equality: execute the final metric, never the saved
expression.
+ entry = [metric_outputs[label], *entry[1:]]
+ retained.append(entry)
+ return retained
+
+
+def _filter_identity(filter_: Any) -> tuple[Any, ...] | None:
+ """Return the native identity used when one filter replaces another."""
+ if not isinstance(filter_, Mapping):
+ return None
+ return (
+ filter_.get("clause"),
+ filter_.get("expressionType"),
+ filter_.get("subject"),
+ filter_.get("operator"),
+ )
+
+
+def _temporal_binding_filter(filters: list[Any], subject: Any) -> dict[str,
Any] | None:
+ """Find the unique filter owned by a recorded MCP temporal marker."""
+ if subject is None:
+ return None
+ if not isinstance(subject, str) or not subject:
+ raise ValueError(
+ "MCP temporal binding provenance subject must be a non-empty
string"
+ )
+ matches = [
+ filter_
+ for filter_ in filters
+ if isinstance(filter_, dict)
+ and filter_.get("subject") == subject
+ and filter_.get("operator") == FilterOperator.TEMPORAL_RANGE.value
+ ]
+ if len(matches) != 1:
+ raise ValueError(
+ "MCP temporal binding provenance must match exactly one "
+ f"TEMPORAL_RANGE filter for subject {subject!r}; found
{len(matches)}"
+ )
+ return matches[0]
+
+
+def _append_or_replace_filter(filters: list[Any], filter_: Any) -> None:
+ """Append a filter, replacing the same native role when identifiable."""
+ identity = _filter_identity(filter_)
+ if identity is None:
+ if filter_ not in filters:
+ filters.append(filter_)
+ return
+ filters[:] = [item for item in filters if _filter_identity(item) !=
identity]
+ filters.append(filter_)
+
+
+_NATIVE_TEMPORAL_ROLE_FIELDS: dict[str, frozenset[str]] = {
+ # Typed ``x`` is persisted as native x_axis/granularity_sqla for XY and
+ # Mixed Timeseries. Waterfall exposes the typed field as ``x_axis``.
+ "x_axis": frozenset({"x", "x_axis"}),
+ "granularity_sqla": frozenset({"x", "x_axis", "temporal_column"}),
+ # Chart plugins may designate a chart-specific query role as the implicit
+ # dashboard-time subject.
+ "start_time": frozenset({"start_time"}),
+}
+
+
+def _native_temporal_subject_changed(
+ existing_form_data: Mapping[str, Any],
+ new_form_data: Mapping[str, Any],
+ explicit_fields: set[str],
+) -> bool:
+ """Return whether an authoritative native temporal role was replaced.
+
+ Mapping a partial update can propose a dataset fallback binding even when
+ the caller only changed filters. That proposal is not authoritative. A
+ changed x/granularity/chart-specific role is authoritative only when its
+ corresponding typed field was actually supplied.
+ """
+ for native_key, typed_fields in _NATIVE_TEMPORAL_ROLE_FIELDS.items():
+ if explicit_fields.isdisjoint(typed_fields):
+ continue
+ existing_value = existing_form_data.get(native_key)
+ incoming_value = new_form_data.get(native_key)
+ if existing_value != incoming_value:
+ return True
+ return False
+
+
+def _native_temporal_binding(
+ form_data: Mapping[str, Any], filters: list[Any]
+) -> tuple[str | None, dict[str, Any] | None]:
+ """Resolve one binding for a trusted native temporal role, if present."""
+ for native_key in _NATIVE_TEMPORAL_ROLE_FIELDS:
+ subject = form_data.get(native_key)
+ if not isinstance(subject, str) or not subject:
+ continue
+ matches = [
+ filter_
+ for filter_ in filters
+ if isinstance(filter_, dict)
+ and filter_.get("subject") == subject
+ and filter_.get("operator") == FilterOperator.TEMPORAL_RANGE.value
+ ]
+ if len(matches) > 1:
+ raise ValueError(
+ "An authoritative native temporal subject must match at most
one "
+ f"TEMPORAL_RANGE filter for subject {subject!r}; found "
+ f"{len(matches)}"
+ )
+ if matches:
+ return subject, matches[0]
+ return None, None
+
+
+def merge_update_form_data( # noqa: C901
+ existing_form_data: Mapping[str, Any],
+ new_form_data: Dict[str, Any],
+ config: ChartConfig,
+) -> None:
+ """Apply the shared omission/provenance contract for chart updates.
+
+ Mapper-generated neutral temporal bindings are infrastructure, not evidence
+ that the caller supplied ``filters`` or changed a saved time-range binding.
+ This helper is used by immediate saves, preview-first saved updates, and
+ cached-preview updates so omission, clear, replacement, and temporal
+ overrides have identical behavior.
+
+ State never crosses a visualization boundary: a viz-type change starts from
+ the mapper's output, so the previous chart's predicates are not restored.
+ """
+ existing_viz_type = existing_form_data.get("viz_type")
+ if isinstance(existing_viz_type, str) and existing_viz_type !=
new_form_data.get(
Review Comment:
A saved chart whose `params` omit `viz_type` (REST-created charts can;
`Slice.form_data` injects it from the column) is treated as a viz change by
`merge_chart_form_data`, which returns the fresh form data, but this guard only
returns for a string `viz_type`, so the old `adhoc_filters` are restored.
Converting a Pie chart saved with a `region = 'US'` filter to a Table with no
`filters` then returns only US rows, and
`BulletChartPlugin.resolve_update_config` likewise ignores the saved hierarchy
for such a chart. Should `_get_existing_form_data` inject `chart.viz_type` (as
`Slice.form_data` does) so both merges agree, or should this guard treat a
missing value like a mismatch the way `merge_same_viz_form_data` does?
##########
superset/mcp_service/chart/chart_helpers.py:
##########
@@ -589,23 +591,518 @@ def resolve_big_number_columns(form_data: dict[str,
Any]) -> list[Any]:
return [granularity] if isinstance(granularity, str) and granularity else
[]
-def resolve_gantt_query_fields( # noqa: C901
- form_data: dict[str, Any],
-) -> tuple[list[Any], list[Any], list[list[Any]], list[Any]]:
- """Mirror the ECharts Gantt ``buildQuery`` field extraction contract.
+def extract_x_axis_col(form_data: dict[str, Any]) -> str | None:
+ """Return the x_axis column name from form_data, or None if not set."""
+ x_axis = form_data.get("x_axis")
+ if isinstance(x_axis, str) and x_axis:
+ return x_axis
+ if isinstance(x_axis, dict):
+ col_name = x_axis.get("column_name")
+ return col_name if isinstance(col_name, str) and col_name else None
+ return None
+
+
+def _x_axis_query_field(form_data: dict[str, Any]) -> Any | None:
+ """Resolve a frontend x-axis value without losing SQL expressions."""
+ x_axis = form_data.get("x_axis")
+ if isinstance(x_axis, str) and x_axis:
+ return x_axis
+ if isinstance(x_axis, dict):
+ if (
+ isinstance(x_axis.get("sqlExpression"), str)
+ and x_axis.get("sqlExpression")
+ and isinstance(x_axis.get("label"), str)
+ and x_axis.get("label")
+ and x_axis.get("expressionType") in (None, "SQL")
+ ):
+ return x_axis
+ column_name = x_axis.get("column_name") or x_axis.get("columnName")
+ if isinstance(column_name, str) and column_name:
+ return column_name
+ return None
+
+
+def _normalized_x_axis_query_field(form_data: dict[str, Any]) -> Any | None:
+ """Mirror ``buildQueryContext.normalizeTimeColumn`` for a set x-axis."""
+ x_axis = _x_axis_query_field(form_data)
+ if x_axis is None:
+ return None
+ time_grain = form_data.get("time_grain_sqla")
+ if isinstance(x_axis, str):
+ normalized = {
+ "columnType": "BASE_AXIS",
+ "sqlExpression": x_axis,
+ "label": x_axis,
+ "expressionType": "SQL",
+ "isColumnReference": True,
+ }
+ if time_grain is not None:
+ normalized["timeGrain"] = time_grain
+ return normalized
+ normalized = {"columnType": "BASE_AXIS", **x_axis}
+ # The original adhoc column's grain overrides the common control, matching
+ # the frontend spread order.
+ if "timeGrain" not in normalized and time_grain is not None:
+ normalized["timeGrain"] = time_grain
+ return normalized
+
+
+def _resolve_big_number_query_columns(form_data: dict[str, Any]) -> list[Any]:
+ """Resolve only Big Number's explicit x-axis query column.
+
+ The frontend keeps ``granularity_sqla`` out of ``columns`` and asks the
+ backend for a timeseries instead, which yields ``__timestamp``. An explicit
+ ``x_axis`` is different: the plugin retains that column in the final query.
+ """
+ if (x_axis := _normalized_x_axis_query_field(form_data)) is not None:
+ return [x_axis]
+ return []
+
+
+def _as_list(value: Any) -> list[Any]:
+ """Match the frontend's ``ensureIsArray`` for query controls."""
+ if value is None:
+ return []
+ return value if isinstance(value, list) else [value]
+
+
+def _column_label(column: Any) -> str | None:
+ """Return the frontend ``getColumnLabel`` value for a query column."""
+ if isinstance(column, str):
+ return column
+ if not isinstance(column, dict):
+ return None
+ return (
+ column.get("label") or column.get("sqlExpression") or
column.get("column_name")
+ )
+
+
+def _metric_label(metric: Any) -> str | None:
+ """Return the frontend ``getMetricLabel`` value for a query metric."""
+ if isinstance(metric, str):
+ return metric
+ if not isinstance(metric, dict):
+ return None
+ if label := metric.get("label"):
+ return label
+ if metric.get("expressionType") == "SIMPLE":
+ column = metric.get("column") or {}
+ name = (
+ column.get("columnName") or column.get("column_name")
+ if isinstance(column, dict)
+ else None
+ )
+ if name and metric.get("aggregate"):
+ return f"{metric['aggregate']}({name})"
+ return metric.get("sqlExpression")
+
+
+def _is_query_form_metric(value: Any) -> bool:
+ """Mirror the frontend's ``isQueryFormMetric`` type guard."""
+ return isinstance(value, str) or (
+ isinstance(value, dict) and value.get("expressionType") in {"SIMPLE",
"SQL"}
+ )
- Returns ``(columns, metrics, orderby, series_columns)``. Saved form data is
- user-editable, so malformed or oversized native ordering is rejected rather
- than silently dropped or passed into ``QueryContextFactory``.
+
+def _timeseries_base_metrics(form_data: dict[str, Any]) -> list[Any]:
+ """Return metrics extracted by the common frontend query-field aliases."""
+ metrics = [*_as_list(form_data.get("metrics")),
*_as_list(form_data.get("metric"))]
+ if (size := form_data.get("size")) is not None:
+ metrics.append(size)
+ return _dedupe_query_fields(metrics, _metric_label)
+
+
+def _timeseries_extra_metrics(form_data: dict[str, Any]) -> list[Any]:
+ """Mirror ``extractExtraMetrics`` for ungrouped x-axis sorting."""
+ if _as_list(form_data.get("groupby")):
+ return []
+ limit_metrics = _as_list(form_data.get("timeseries_limit_metric"))
+ if not limit_metrics:
+ return []
+ limit_metric = limit_metrics[0]
+ limit_label = _metric_label(limit_metric)
+ if not limit_label or limit_label != form_data.get("x_axis_sort"):
+ return []
+ if any(
+ _metric_label(metric) == form_data.get("x_axis_sort")
+ for metric in _as_list(form_data.get("metrics"))
+ ):
+ return []
+ return [limit_metric]
+
+
+def _query_series_columns(query: dict[str, Any]) -> list[Any]:
+ """Resolve pivot columns with JavaScript's array-truthiness semantics.
+
+ JavaScript treats an explicitly empty array as truthy, so
+ ``series_columns: []`` must not fall through to the query's x-axis column.
+ Only an absent/null series-columns field falls back to ``columns``.
"""
- from superset.utils import json as utils_json
+ if "series_columns" in query and query["series_columns"] is not None:
+ return _as_list(query["series_columns"])
+ return _as_list(query.get("columns"))
+
+
+def _dedupe_query_fields(values: list[Any], labeler: Any) -> list[Any]:
+ """Preserve the first query field for each frontend result label."""
+ result: list[Any] = []
+ seen: set[str] = set()
+ for value in values:
+ label = labeler(value)
+ if not label or label in seen:
+ continue
+ seen.add(label)
+ result.append(value)
+ return result
+
+
+def _deck_tooltip_columns(value: Any) -> list[str]:
+ """Extract the physical tooltip fields accepted by Deck.gl plugins."""
+ if not isinstance(value, list):
+ return []
+ columns: list[str] = []
+ for item in value:
+ column: Any = None
+ if isinstance(item, str):
+ column = item
+ elif isinstance(item, dict) and item.get("item_type") == "column":
+ column = item.get("column_name")
+ if isinstance(column, str) and column and column not in columns:
+ columns.append(column)
+ return columns
+
+
+def _deck_base_query_fields( # noqa: C901
+ form_data: dict[str, Any],
+) -> tuple[list[Any], list[Any], list[list[Any]] | None]:
+ """Mirror the common fields built before a Deck.gl layer adapter runs."""
+ query_mode = form_data.get("query_mode")
+ columns: list[Any] = []
+ metrics: list[Any] = []
+ raw_orderby: list[Any] = []
+ aliases = {
+ "metric": "metrics",
+ "metric_2": "metrics",
+ "secondary_metric": "metrics",
+ "left_metric": "metrics",
+ "right_metric": "metrics",
+ "x": "metrics",
+ "y": "metrics",
+ "size": "metrics",
+ "all_columns": "columns",
+ "series": "groupby",
+ "order_by_cols": "orderby",
+ }
+ for key, value in form_data.items():
+ if value is None:
+ continue
+ normalized = aliases.get(key, key)
+ if query_mode == "aggregate" and normalized == "columns":
+ continue
+ if query_mode == "raw" and normalized in {"groupby", "metrics"}:
+ continue
+ if normalized == "groupby":
+ normalized = "columns"
+ if normalized == "columns":
+ columns.extend(_as_list(value))
+ elif normalized == "metrics":
+ metrics.extend(_as_list(value))
+ elif normalized == "orderby":
+ raw_orderby.extend(_as_list(value))
+
+ orderby: list[list[Any]] = []
+ if len(raw_orderby) > 100:
+ raise ValueError("Deck orderby must contain at most 100 entries")
+ for index, value in enumerate(raw_orderby):
+ if isinstance(value, str):
+ if len(value) > 1000:
+ raise ValueError(f"Deck orderby[{index}] is too long")
+ try:
+ value = utils_json.loads(value)
+ except (TypeError, ValueError) as ex:
+ raise ValueError(f"Deck orderby[{index}] is not valid JSON")
from ex
+ if (
+ not isinstance(value, (list, tuple))
+ or len(value) != 2
+ or not isinstance(value[1], bool)
+ ):
+ raise ValueError(
+ f"Deck orderby[{index}] must be [field, ascending_boolean]"
+ )
+ orderby.append(list(value))
+
+ return (
+ _dedupe_query_fields(columns, _column_label),
+ _dedupe_query_fields(metrics, _metric_label),
+ orderby or None,
+ )
+
+
+def _deck_add_columns(columns: list[Any], *values: Any) -> list[Any]:
+ """Add layer fields by frontend result label without duplicates."""
+ expanded = list(columns)
+ for value in values:
+ expanded.extend(_as_list(value))
+ return _dedupe_query_fields(expanded, _column_label)
+
+
+def _deck_add_metrics(metrics: list[Any], *values: Any) -> list[Any]:
+ """Add layer metric roles without emitting the same output label twice."""
+ expanded = list(metrics)
+ for value in values:
+ expanded.extend(_as_list(value))
+ return _dedupe_query_fields(expanded, _metric_label)
+
+
+def _deck_fixed_or_metric(
+ value: Any, *, allow_legacy_string: bool = True
+) -> Any | None:
+ """Return the metric stored in a Deck fixed-or-metric control."""
+ if allow_legacy_string and isinstance(value, str) and value:
+ # Legacy Deck controls store the metric key directly as a string.
+ return value
+ if not isinstance(value, dict) or value.get("type") != "metric":
+ return None
+ metric = value.get("value")
+ return (
+ metric if metric is not None and not isinstance(metric, (int, float))
else None
+ )
+
+
+def _deck_spatial_columns(value: Any) -> list[str]:
+ """Resolve one complete frontend Deck spatial configuration."""
+ if not isinstance(value, dict) or not value.get("type"):
+ raise ValueError("Bad spatial key")
+ spatial_type = value["type"]
+ fields = {
+ "latlong": ("lonCol", "latCol"),
+ "delimited": ("lonlatCol",),
+ "geohash": ("geohashCol",),
+ }.get(spatial_type)
+ if fields is None:
+ raise ValueError(f"Unknown spatial type: {spatial_type}")
+ columns: list[str] = []
+ for field in fields:
+ column = value.get(field)
+ if not isinstance(column, str) or not column:
+ raise ValueError(f"Incomplete {spatial_type} spatial
configuration")
+ columns.append(column)
+ return columns
+
+
+def _deck_add_null_filters(
+ query: dict[str, Any], columns: list[str], *, include_null_value: bool =
False
+) -> None:
+ """Append renderer-required non-null filters, deduplicated by column."""
+ filters = list(query.get("filters") or [])
+ present = {
+ item.get("col")
+ for item in filters
+ if isinstance(item, dict) and item.get("op") == "IS NOT NULL"
+ }
+ for column in columns:
+ if not column or column in present:
+ continue
+ filter_: dict[str, Any] = {"col": column, "op": "IS NOT NULL"}
+ if include_null_value:
+ filter_["val"] = None
+ filters.append(filter_)
+ present.add(column)
+ query["filters"] = filters
+
+
+def _build_deck_query( # noqa: C901
Review Comment:
With Deck queries now built here, `resolve_deck_gl_columns`,
`_resolve_deck_gl_metrics` and `_deck_gl_null_filters` (and their helpers
`_deck_gl_spatial_cols` and `_is_metric_ref`) have no production caller, yet
`test_chart_helpers.py` still exercises them. A later fix to Scatter
tooltip-column selection could land in `resolve_deck_gl_columns`, pass its
tests, and change no MCP query. Can they and their tests be removed?
##########
superset/mcp_service/chart/query_result.py:
##########
@@ -18,90 +18,2066 @@
"""Helpers for interpreting ChartDataCommand result envelopes."""
import math
-from collections.abc import Mapping
-from decimal import Decimal
+import re
+import time as system_time
+from bisect import bisect_right
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import date, datetime, time, timedelta, timezone
+from decimal import Decimal, InvalidOperation
+from enum import Enum
from numbers import Real
+from types import MappingProxyType
from typing import Any, cast
+from uuid import UUID
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+import numpy as np
+import pandas as pd
+import pytz
+from dateutil import tz as dateutil_tz
+from dateutil.tz.tz import _ttinfo as dateutil_ttinfo
+from dateutil.zoneinfo import tzfile as dateutil_zoneinfo_tzfile
+from pydantic import BaseModel
+from pydantic_core import to_json
from superset.mcp_service.chart.schemas import ChartError
+from superset.mcp_service.utils.serialization import decode_binary
+from superset.utils.core import GenericDataType
+from superset.utils.dates import datetime_to_epoch, EPOCH
FAILED_QUERY_STATUSES = frozenset(
{"error", "failed", "stopped", "timed_out", "cancelled", "canceled"}
)
+_ERROR_KEYS = ("error", "error_message", "message", "detail")
+_MAX_ERROR_DEPTH = 32
+_MAX_ERROR_ITEMS = 256
+_MAX_SEQUENCE_ITEMS = 64
+_MAX_ERROR_PARTS = 3
+_MAX_ERROR_BYTES = 2000
+_MAX_INTEGER_DIGITS = 1000
+_MAX_QUERY_COUNT = 64
+_MAX_QUERY_COLUMNS = 4096
+_MAX_COLUMN_NAME_BYTES = 4096
+_MAX_ROW_CONTAINER_DEPTH = 32
+_MAX_ROW_CONTAINER_ITEMS = 4096
+_MAX_CACHE_STRING_BYTES = 4096
+_MAX_RESULT_ROW_COUNT = (1 << 63) - 1
-def _query_error_text(value: Any) -> str | None:
- """Convert a bounded query error payload into a useful message."""
- if value is None or value is False:
+# Chart results are routinely much larger than an MCP response should return,
but
+# legitimate exports and high-cardinality chart queries still need useful room.
+# Each query may return Superset's configured 50k ROW_LIMIT. The aggregate row
+# budget admits both legs of Big Number raw/trend and Mixed Timeseries results
+# at that limit, while the value budget admits twenty scalar columns on both
+# legs (plus their row containers). The complete compact JSON projection is
+# capped at 16 MiB, including scalar tokens, escaping, keys, and syntax.
Metadata
+# profiling has a separate row-by-column work budget in ``response_utils`` so
+# wide sparse results cannot turn bounded validation into an unbounded scan.
+# Individual source-result cell strings are capped at 64 KiB and object keys at
+# 4 KiB. Derived strings in a final Pydantic response have no per-cell cap; the
+# complete compact response remains subject to the 16 MiB aggregate budget.
+# Query metadata has its own 1 MiB aggregate budget so SQL and cache metadata
+# cannot consume the row-data allowance. Row-shaped indexnames use the row-data
+# work budget while retaining the metadata byte budget. Integer/Decimal bounds
+# prevent later hashing, uniqueness, and JSON conversion from allocating by
magnitude.
+MAX_QUERY_RESULT_ROWS = 50_000
+MAX_QUERY_RESULT_TOTAL_ROWS = 2 * MAX_QUERY_RESULT_ROWS
+MAX_QUERY_RESULT_VALUES = 2_500_000
+MAX_QUERY_RESULT_VALUE_BYTES = 16 * 1024 * 1024
+MAX_QUERY_RESULT_METADATA_BYTES = 1024 * 1024
+MAX_QUERY_RESULT_METADATA_ITEMS = 32_768
+MAX_QUERY_RESULT_WORK = MAX_QUERY_RESULT_VALUES +
MAX_QUERY_RESULT_METADATA_ITEMS
+MAX_QUERY_RESULT_STRING_BYTES = 64 * 1024
+MAX_QUERY_RESULT_KEY_BYTES = 4096
+MAX_QUERY_RESULT_INTEGER_BITS = 4096
+MAX_QUERY_RESULT_INTEGER_DIGITS = 1234
+MAX_QUERY_RESULT_DECIMAL_DIGITS = 1024
+MAX_QUERY_RESULT_DECIMAL_EXPONENT = 4096
+MAX_QUERY_RESULT_DECIMAL_STORAGE = 2048
+_BUILTIN_SCALAR_TYPES = (str, bytes, bytearray, memoryview, int, float, bool)
+_SCALAR_BASE_TYPES = (*_BUILTIN_SCALAR_TYPES, Enum)
+_SUPPORTED_COLTYPES = frozenset(GenericDataType)
+_TRUSTED_TZINFO_TYPES = (timezone, ZoneInfo)
+_DATEUTIL_TZFILE_TYPE = dateutil_tz.tzfile
+_DATEUTIL_TZOFFSET_TYPE = type(dateutil_tz.tzoffset(None, 0))
+_DATEUTIL_TZUTC_TYPE = type(dateutil_tz.UTC)
+_DATEUTIL_TZLOCAL_TYPE = type(dateutil_tz.tzlocal())
+_MAX_DATEUTIL_TRANSITIONS = 4096
+_PYTZ_FIXED_OFFSET_TYPE = type(pytz.FixedOffset(1))
+_PYTZ_UTC_TYPE = type(pytz.UTC)
+_PYTZ_NAMED_BASE_TYPES = (pytz.tzinfo.DstTzInfo, pytz.tzinfo.StaticTzInfo)
+_NUMPY_INTEGER_TYPES = frozenset(
+ type(value)
+ for value in (
+ np.int8(0),
+ np.int16(0),
+ np.int32(0),
+ np.int64(0),
+ np.uint8(0),
+ np.uint16(0),
+ np.uint32(0),
+ np.uint64(0),
+ )
+)
+_NUMPY_FLOAT_TYPES = frozenset(
+ type(value)
+ for value in (np.float16(0), np.float32(0), np.float64(0),
np.longdouble(0))
+)
+_NUMPY_EXTENDED_FLOAT_TYPES = frozenset(
+ type_
+ for type_ in _NUMPY_FLOAT_TYPES
+ if np.finfo(type_).nmant > np.finfo(np.float64).nmant
+)
+_PANDAS_NAT_TYPE = type(pd.NaT)
+_PANDAS_NA_TYPE = type(pd.NA)
+_PANDAS_PERIOD_TYPE = type(pd.Period("2000-01", freq="M"))
+_PANDAS_INTERVAL_TYPE = type(pd.Interval(0, 1))
+_EXCEL_MIN_DATE = date(1900, 1, 1)
+# Leave room for XLSX readers' millisecond rounding at the final serial day.
+_EXCEL_MAX_TIME = time(23, 59, 59, 999000)
+_EXCEL_MAX_DATETIME = datetime.combine(date.max, _EXCEL_MAX_TIME)
+# Serial day 2958465 is 9999-12-31 in Excel's 1900 date system.
+_EXCEL_MAX_DURATION = timedelta(
+ days=2958465, hours=23, minutes=59, seconds=59, milliseconds=999
+)
+
+
+@dataclass(frozen=True)
+class _ErrorText:
+ """Bounded error extraction outcome."""
+
+ text: str | None = None
+ malformed: str | None = None
+
+
+@dataclass
+class _ResultBudget:
+ """Aggregate work counters shared across all queries in one result."""
+
+ rows: int = 0
+ values: int = 0
+ json_bytes: int = 0
+ metadata_items: int = 0
+ metadata_bytes: int = 0
+
+
+def _truncate_utf8(value: str, max_bytes: int) -> str:
+ """Return bounded, replacement-decoded UTF-8 text.
+
+ Encoding even the non-truncated path is intentional: Python strings may
+ contain unpaired surrogates, while MCP/JSON responses must always be valid
+ UTF-8. Slicing by characters before encoding also prevents an
+ attacker-sized string from being encoded in full.
+ """
+ if max_bytes <= 0:
+ return ""
+ candidate = value[:max_bytes]
+ encoded = candidate.encode("utf-8", errors="replace")
+ if len(encoded) <= max_bytes and len(candidate) == len(value):
+ return encoded.decode("utf-8", errors="replace")
+ suffix = "... [truncated]"
+ suffix_bytes = suffix.encode()
+ if max_bytes <= len(suffix_bytes):
+ return suffix_bytes[:max_bytes].decode("ascii")
+ content_limit = max(0, max_bytes - len(suffix_bytes))
+ content = encoded[:content_limit].decode("utf-8", errors="ignore")
+ return content + suffix
+
+
+def _type_descriptor(value: Any, max_bytes: int) -> str | None:
+ """Describe an unsupported value without consulting its implementation."""
+ if max_bytes <= 0:
return None
- if isinstance(value, Mapping):
- for key in ("error", "error_message", "message", "detail"):
- if text := _query_error_text(value.get(key)):
- return text
+ value_type = type(value)
+ try:
+ type_name = type.__getattribute__(value_type, "__name__")
+ except (AttributeError, TypeError): # pragma: no cover - defensive
metaclass
+ type_name = "unknown"
+ if type(type_name) is not str:
+ type_name = "unknown"
+ bounded_name = _truncate_utf8(type_name, max_bytes)
+ return _truncate_utf8(f"<{bounded_name} object>", max_bytes)
+
+
+def _type_mro(value_type: type[Any]) -> tuple[type[Any], ...]:
+ """Read a concrete type's MRO without consulting its metaclass
overrides."""
+ try:
+ mro = type.__getattribute__(value_type, "__mro__")
+ except (AttributeError, TypeError): # pragma: no cover - all normal types
have MRO
+ return ()
+ return mro if type(mro) is tuple else ()
+
+
+def _mro_contains(
+ value_mro: tuple[type[Any], ...], base_types: tuple[type[Any], ...]
+) -> bool:
+ """Return whether an MRO contains a base, using identity-only
comparisons."""
+ return any(
+ base is expected_base for base in value_mro for expected_base in
base_types
+ )
+
+
+def _safe_scalar_text(value: Any, max_bytes: int) -> str | None: # noqa: C901
+ """Render a bounded scalar without invoking attacker-controlled string
code."""
+ value_type = type(value)
+ if _mro_contains(_type_mro(value_type), (Enum,)):
+ try:
+ enum_value = object.__getattribute__(value, "_value_")
+ except Exception:
+ return _type_descriptor(value, max_bytes)
+ if not any(
+ type(enum_value) is scalar_type for scalar_type in
_BUILTIN_SCALAR_TYPES
+ ):
+ return _type_descriptor(value, max_bytes)
+ return _safe_scalar_text(enum_value, max_bytes)
+ if value is None or value is False:
return None
- if isinstance(value, (list, tuple)):
- parts = [text for item in value if (text := _query_error_text(item))]
- return "; ".join(parts[:3]) or None
- text = str(value)
- return text[:2000] if text else None
+ if value_type is str:
+ return _truncate_utf8(value, max_bytes) if value else None
+ if value_type is bytes or value_type is bytearray or value_type is
memoryview:
+ try:
+ view = memoryview(value).cast("B")
+ sample = view[: max(0, max_bytes)].tobytes()
+ text = sample.decode("utf-8", errors="replace")
+ if len(view) > len(sample):
+ text += "... [truncated]"
+ return _truncate_utf8(text, max_bytes) if text else None
+ except (TypeError, ValueError):
+ return _type_descriptor(value, max_bytes)
+ if value_type is int:
+ digits = (
+ 1 if value == 0 else int((abs(value).bit_length() - 1) *
math.log10(2)) + 1
+ )
+ if digits > _MAX_INTEGER_DIGITS:
+ sign = "negative " if value < 0 else ""
+ return _truncate_utf8(
+ f"<{sign}integer with approximately {digits} decimal digits>",
+ max_bytes,
+ )
+ return _truncate_utf8(str(value), max_bytes)
+ if value_type is bool or value_type is float:
+ return _truncate_utf8(str(value), max_bytes)
+ return _type_descriptor(value, max_bytes)
+
+
+def _query_error_text(value: Any) -> _ErrorText: # noqa: C901
+ """Iteratively extract actionable text from an untrusted error payload.
+
+ Chart backends and engine adapters can return arbitrary nested error
shapes.
+ Depth, visited-item, sequence-width, and output-byte limits keep validation
+ deterministic even for cycles, repeated containers, and adversarial values.
+ """
+ stack: list[tuple[Any, int]] = [(value, 0)]
+ seen: set[int] = set()
+ parts: list[str] = []
+ visited = 0
+ used_bytes = 0
+ while stack and len(parts) < _MAX_ERROR_PARTS:
+ item, depth = stack.pop()
+ visited += 1
+ if visited > _MAX_ERROR_ITEMS:
+ return _ErrorText(malformed="error payload exceeds the item limit")
+ if depth > _MAX_ERROR_DEPTH:
+ return _ErrorText(malformed="error payload exceeds the depth
limit")
-def _failure_for_query_payload(
- payload: Mapping[str, Any], label: str
+ # ChartDataCommand envelopes cross a JSON boundary. Only exact JSON
+ # containers are trusted here: ABC/isinstance checks can consult a
+ # spoofed ``__class__``, and subclass get/contains/iter/len hooks are
+ # attacker-controlled. Exact dict/list operations below are builtin and
+ # non-overridable.
+ is_mapping = type(item) is dict
+ is_sequence = type(item) is list
+ item_mro = _type_mro(type(item))
+ if not (is_mapping or is_sequence) and (
+ _mro_contains(item_mro, (dict, list, Mapping, Sequence))
+ and not _mro_contains(item_mro, _SCALAR_BASE_TYPES)
+ ):
+ return _ErrorText(
+ malformed="error payload contains an unsupported container
type"
+ )
+ if is_mapping or is_sequence:
+ identity = id(item)
+ if identity in seen:
+ return _ErrorText(
+ malformed="error payload contains repeated or cyclic
containers"
+ )
+ seen.add(identity)
+
+ if is_mapping:
+ children: list[Any] = []
+ for key in _ERROR_KEYS:
+ if dict.__contains__(item, key):
+ children.append(dict.__getitem__(item, key))
+ if not children:
+ if dict.__len__(item):
+ return _ErrorText(
+ malformed=(
+ "error payload object has no recognized message
field"
+ )
+ )
+ stack.extend((child, depth + 1) for child in reversed(children))
+ continue
+
+ if is_sequence:
+ width = list.__len__(item)
+ if width > _MAX_SEQUENCE_ITEMS:
+ return _ErrorText(malformed="error payload exceeds the width
limit")
+ children = [list.__getitem__(item, index) for index in
range(width)]
+ stack.extend((child, depth + 1) for child in reversed(children))
+ continue
+
+ remaining = _MAX_ERROR_BYTES - used_bytes - (2 if parts else 0)
+ text = _safe_scalar_text(item, remaining)
+ if text:
+ parts.append(text)
+ used_bytes += len(text.encode("utf-8", errors="replace")) + (
+ 2 if len(parts) > 1 else 0
+ )
+
+ return _ErrorText(text="; ".join(parts) or None)
+
+
+def _failure_for_query_payload( # noqa: C901
+ payload: dict[str, Any], label: str
) -> ChartError | None:
"""Extract one failure from a top-level or per-query payload."""
+ malformed: str | None = None
for key in ("error", "errors", "error_message"):
- if message := _query_error_text(payload.get(key)):
+ extracted = _query_error_text(dict.get(payload, key))
+ if extracted.malformed:
+ malformed = malformed or extracted.malformed
+ continue
+ if message := extracted.text:
return ChartError(
error=f"{label} failed: {message}", error_type="QueryError"
)
- raw_status = payload.get("status")
- status = str(getattr(raw_status, "value", raw_status) or "")
+ raw_status = dict.get(payload, "status")
+ status = _safe_scalar_text(raw_status, 200) or ""
normalized_status = status.strip().casefold().replace("-", "_").replace("
", "_")
if normalized_status in FAILED_QUERY_STATUSES:
- message = (
- _query_error_text(payload.get("message"))
- or _query_error_text(payload.get("error_message"))
- or normalized_status
- )
+ extracted = _query_error_text(dict.get(payload, "message"))
+ if extracted.malformed:
+ malformed = malformed or extracted.malformed
+ fallback = _query_error_text(dict.get(payload, "error_message"))
+ if fallback.malformed:
+ malformed = malformed or fallback.malformed
+ if malformed and not (extracted.text or fallback.text):
+ return _malformed_result(malformed)
+ message = extracted.text or fallback.text or normalized_status
return ChartError(error=f"{label} failed: {message}",
error_type="QueryError")
- if payload.get("success") is False:
- message = _query_error_text(payload.get("message")) or "request failed"
+ if dict.get(payload, "success") is False:
+ extracted = _query_error_text(dict.get(payload, "message"))
+ if extracted.malformed:
+ malformed = malformed or extracted.malformed
+ if malformed and not extracted.text:
+ return _malformed_result(malformed)
+ message = extracted.text or "request failed"
return ChartError(error=f"{label} failed: {message}",
error_type="QueryError")
if (
raw_status is None
- and "data" not in payload
- and "queries" not in payload
- and (message := _query_error_text(payload.get("message")))
+ and "data" not in dict.keys(payload)
+ and "queries" not in dict.keys(payload)
):
- return ChartError(error=f"{label} failed: {message}",
error_type="QueryError")
+ extracted = _query_error_text(dict.get(payload, "message"))
+ if extracted.malformed:
+ malformed = malformed or extracted.malformed
+ if extracted.text:
+ return ChartError(
+ error=f"{label} failed: {extracted.text}",
error_type="QueryError"
+ )
+ if malformed:
+ return _malformed_result(malformed)
return None
-def query_result_failure(result: Any) -> ChartError | None:
- """Return a structured failure embedded in a ChartDataCommand payload.
+def _malformed_result(message: str) -> ChartError:
+ """Build a stable error for an invalid ChartDataCommand envelope."""
+ return ChartError(
+ error=f"Malformed chart query result: {message}",
+ error_type="MalformedQueryResult",
+ )
- ChartDataCommand can return an HTTP-successful envelope whose top level or
- any query reports a failure. Every query is inspected before callers accept
- data from the result. Successful statuses may carry informational messages,
- so ``message`` alone is not treated as an error.
+
+def bounded_result_row_count(value: Any) -> int | None:
+ """Return one exact bounded row count, rejecting coercive lookalikes."""
+ if value is None:
+ return None
+ if type(value) is int:
+ count = value
+ elif type(value) is float and math.isfinite(value) and value.is_integer():
+ count = int(value)
+ else:
+ raise ValueError("must be a finite non-negative integral number")
+ if count < 0:
+ raise ValueError("must be non-negative")
+ if count > _MAX_RESULT_ROW_COUNT:
+ raise ValueError("exceeds the supported bound")
+ return count
+
+
+def _bounded_utf8_length(value: str, max_bytes: int) -> int | None:
+ """Return an exact UTF-8 size without encoding attacker-sized text."""
+ if str.__len__(value) > max_bytes:
+ return None
+ try:
+ encoded = str.encode(value, "utf-8", errors="strict")
+ except UnicodeEncodeError:
+ return None
+ size = bytes.__len__(encoded)
+ return size if size <= max_bytes else None
+
+
+def _json_string_size(value: str, max_bytes: int) -> int | None:
+ """Return the exact UTF-8 size of a JSON string without serializing it."""
+ raw_size = _bounded_utf8_length(value, max_bytes)
+ if raw_size is None:
+ return None
+ escaped_size = raw_size + 2 # surrounding quotes
+ for character in value:
+ codepoint = ord(character)
+ if character in {'"', "\\"} or character in {"\b", "\t", "\n", "\f",
"\r"}:
+ escaped_size += 1
+ elif codepoint < 0x20:
+ # Other JSON control characters use a six-byte ``\\u00xx`` escape.
+ escaped_size += 5
+ return escaped_size
+
+
+def _integer_json_size(value: int) -> int:
+ """Return an exact integer JSON size without creating its decimal
string."""
+ magnitude = -value if value < 0 else value
+ if magnitude == 0:
+ digits = 1
+ else:
+ bits = int.bit_length(magnitude)
+ # This fixed-point log10(2) estimate is at most one digit low. Refine
it
+ # with one bounded integer comparison rather than rendering the value.
+ digits = ((bits - 1) * 30103) // 100000 + 1
+ if magnitude >= 10**digits:
+ digits += 1
+ return digits + (value < 0)
+
+
+def _container_json_syntax_size(item_count: int, *, mapping: bool) -> int:
+ """Return braces/brackets, separators, and mapping-colon byte cost."""
+ if item_count == 0:
+ return 2
+ return 2 + item_count - 1 + (item_count if mapping else 0)
+
+
+def _trusted_timedelta_text(value: timedelta) -> str:
+ """Render an exact timedelta with Pydantic's stable ISO-8601 spelling."""
+ total_microseconds = (
+ value.days * 86_400 + value.seconds
+ ) * 1_000_000 + value.microseconds
+ sign = "-" if total_microseconds < 0 else ""
+ remaining = abs(total_microseconds)
+ days, remaining = divmod(remaining, 86_400 * 1_000_000)
+ years, days = divmod(days, 365)
+ hours, remaining = divmod(remaining, 3_600 * 1_000_000)
+ minutes, remaining = divmod(remaining, 60 * 1_000_000)
+ seconds, microseconds = divmod(remaining, 1_000_000)
+
+ date_parts = [f"{years}Y" if years else "", f"{days}D" if days else ""]
+ time_parts = [f"{hours}H" if hours else "", f"{minutes}M" if minutes else
""]
+ if microseconds:
+ fraction = f"{microseconds:06d}".rstrip("0")
+ time_parts.append(f"{seconds}.{fraction}S")
+ elif seconds:
+ time_parts.append(f"{seconds}S")
+
+ date_text = "".join(date_parts)
+ time_text = "".join(time_parts)
+ if not date_text and not time_text:
+ time_text = "0S"
+ return f"{sign}P{date_text}{'T' if time_text else ''}{time_text}"
+
+
+def _chart_data_builtin_timedelta_text(value: timedelta) -> str:
+ """Reproduce ``format_timedelta`` without comparison or string hooks."""
+ total_microseconds = (
+ value.days * 86_400 + value.seconds
+ ) * 1_000_000 + value.microseconds
+ sign = "-" if total_microseconds < 0 else ""
+ remaining = abs(total_microseconds)
+ days, remaining = divmod(remaining, 86_400 * 1_000_000)
+ hours, remaining = divmod(remaining, 3_600 * 1_000_000)
+ minutes, remaining = divmod(remaining, 60 * 1_000_000)
+ seconds, microseconds = divmod(remaining, 1_000_000)
+ day_text = f"{days} {'day' if days == 1 else 'days'}, " if days else ""
+ fraction = f".{microseconds:06d}" if microseconds else ""
+ return f"{sign}{day_text}{hours}:{minutes:02d}:{seconds:02d}{fraction}"
+
+
+def _chart_data_pandas_timedelta_text(value: pd.Timedelta) -> str:
+ """Reproduce Chart Data ``format_timedelta`` output from exact fields."""
+ total_nanoseconds = (
+ (
+ object.__getattribute__(value, "days") * 86_400
+ + object.__getattribute__(value, "seconds")
+ )
+ * 1_000_000
+ + object.__getattribute__(value, "microseconds")
+ ) * 1_000 + object.__getattribute__(value, "nanoseconds")
+ sign = "-" if total_nanoseconds < 0 else ""
+ remaining = abs(total_nanoseconds)
+ days, remaining = divmod(remaining, 86_400 * 1_000_000_000)
+ hours, remaining = divmod(remaining, 3_600 * 1_000_000_000)
+ minutes, remaining = divmod(remaining, 60 * 1_000_000_000)
+ seconds, nanoseconds = divmod(remaining, 1_000_000_000)
+ if nanoseconds % 1_000:
+ fraction = f".{nanoseconds:09d}"
+ elif nanoseconds:
+ fraction = f".{nanoseconds // 1_000:06d}"
+ else:
+ fraction = ""
+ return f"{sign}{days} days
{hours:02d}:{minutes:02d}:{seconds:02d}{fraction}"
+
+
+def _normalized_scalar_json_size( # noqa: C901
+ value: Any, *, max_string_bytes: int = MAX_QUERY_RESULT_STRING_BYTES
+) -> int:
+ """Return a conservative encoded size for one normalized exact scalar."""
+ value_type = type(value)
+ if value is None:
+ return 4
+ if value_type is bool:
+ return 4 if value else 5
+ if value_type is str:
+ size = _json_string_size(value, max_string_bytes)
+ assert size is not None # scalar normalization already bounded the
string
+ return size
+ if value_type is int:
+ return _integer_json_size(value)
+ if value_type is float:
+ if not math.isfinite(value):
+ # Raw Gauge exports retain these markers; JSON responses use null.
+ return 4
+ # Exact builtin repr is hook-free, bounded to a shortest-round-trip
+ # spelling, and avoids pessimistically charging 24 bytes for values
+ # such as 0.0 across ordinary large numeric datasets.
+ return len(float.__repr__(value))
+ if value_type is Decimal:
+ # Decimal storage, coefficient digits, and exponent are bounded before
+ # this point. Its canonical spelling is therefore itself bounded, and
+ # Pydantic serializes Decimal values as JSON strings.
+ text = Decimal.__str__(value)
+ size = _json_string_size(text, MAX_QUERY_RESULT_STRING_BYTES)
+ assert size is not None
+ return size
+ if value_type is datetime:
+ return 40
+ if value_type is date:
+ text = date.isoformat(value)
+ elif value_type is time:
+ return 32
+ elif value_type is timedelta:
+ text = _trusted_timedelta_text(value)
+ elif value_type is UUID:
+ text = UUID.__str__(value)
+ else:
+ raise AssertionError(f"unaccounted normalized scalar: {value_type!r}")
+ size = _json_string_size(text, MAX_QUERY_RESULT_STRING_BYTES)
+ assert size is not None
+ return size
+
+
+def _pydantic_scalar_json_size(value: Any) -> int:
+ """Return the exact Pydantic wire size for a normalized scalar.
+
+ Source-result accounting deliberately retains its existing conservative
+ scalar rules. Final response projections, however, must match
+ pydantic-core's JSON number spelling: for example, it emits ``0.00001`` for
+ ``1e-5`` and ``1e-6`` for ``1e-6`` rather than Python's repr spellings.
"""
- if not isinstance(result, Mapping):
+ if type(value) is float:
+ return len(to_json(value))
+ return _normalized_scalar_json_size(value)
+
+
+def _charge_json_bytes(
+ budget: _ResultBudget, size: int, *, metadata: bool = False
+) -> str | None:
+ """Charge aggregate response bytes and the independent metadata
allowance."""
+ budget.json_bytes += size
+ if budget.json_bytes > MAX_QUERY_RESULT_VALUE_BYTES:
+ return "exceeds the total JSON-encoded byte limit"
+ if metadata:
+ budget.metadata_bytes += size
+ if budget.metadata_bytes > MAX_QUERY_RESULT_METADATA_BYTES:
+ return "metadata exceeds the total JSON-encoded byte limit"
+ return None
+
+
+def _integer_failure(value: int) -> str | None:
+ """Validate exact integer magnitude before decimal rendering or hashing."""
+ bits = int.bit_length(value)
+ if bits > MAX_QUERY_RESULT_INTEGER_BITS:
+ return "contains an integer exceeding the bit-length limit"
+ digits = 1 if bits == 0 else ((bits - 1) * 30103) // 100000 + 1
+ if digits > MAX_QUERY_RESULT_INTEGER_DIGITS:
+ return "contains an integer exceeding the digit limit"
+ return None
+
+
+def _decimal_failure(value: Decimal) -> str | None:
+ """Validate exact Decimal storage, finiteness, digits, and exponent."""
+ if Decimal.__sizeof__(value) > MAX_QUERY_RESULT_DECIMAL_STORAGE:
+ return "contains a Decimal exceeding the storage limit"
+ if not Decimal.is_finite(value):
+ return "contains a non-finite Decimal"
+ parts = Decimal.as_tuple(value)
+ if tuple.__len__(parts.digits) > MAX_QUERY_RESULT_DECIMAL_DIGITS:
+ return "contains a Decimal exceeding the digit limit"
+ exponent = parts.exponent
+ if type(exponent) is not int or abs(exponent) >
MAX_QUERY_RESULT_DECIMAL_EXPONENT:
+ return "contains a Decimal exceeding the exponent limit"
+ return None
+
+
+def _exact_object_namespace(value: Any) -> dict[str, Any] | None:
+ """Read an object's concrete storage without descriptor dispatch."""
+ try:
+ namespace = object.__getattribute__(value, "__dict__")
+ except (AttributeError, TypeError):
+ return None
+ return namespace if type(namespace) is dict else None
+
+
+def _dateutil_timezone_name_without_hooks(tzinfo: Any) -> str | None:
+ """Read a dateutil tzfile's IANA name from exact internal storage."""
+ tzinfo_type = type(tzinfo)
+ if tzinfo_type not in {_DATEUTIL_TZFILE_TYPE, dateutil_zoneinfo_tzfile}:
+ return None
+ namespace = _exact_object_namespace(tzinfo)
+ if namespace is None:
+ return None
+ filename = dict.get(namespace, "_filename")
+ if type(filename) is not str or _bounded_utf8_length(filename, 4096) is
None:
+ return None
+ if tzinfo_type is dateutil_zoneinfo_tzfile:
+ name = filename
+ else:
+ marker = "/zoneinfo/"
+ marker_offset = str.find(filename, marker)
+ if marker_offset >= 0:
+ name = str.__getitem__(filename, slice(marker_offset +
len(marker), None))
+ elif not str.startswith(filename, "/") and str.find(filename, "\\") <
0:
+ name = filename
+ else:
+ return None
+ parts = str.split(name, "/")
+ if not parts or any(part in {"", ".", ".."} for part in parts):
+ return None
+ return name if _bounded_utf8_length(name, 256) is not None else None
+
+
+def _dateutil_ttinfo_without_hooks(
+ value: Any,
+) -> tuple[int, timedelta] | None:
+ """Read one exact dateutil transition record without user-hook dispatch."""
+ if type(value) is not dateutil_ttinfo:
+ return None
+ try:
+ offset = object.__getattribute__(value, "offset")
+ delta = object.__getattribute__(value, "delta")
+ except (AttributeError, TypeError):
+ return None
+ if type(offset) is not int or type(delta) is not timedelta:
+ return None
+ try:
+ if delta != timedelta(seconds=offset):
+ return None
+ except OverflowError:
+ return None
+ return offset, delta
+
+
+def _dateutil_named_offset_without_hooks( # noqa: C901
+ value: datetime, tzinfo: Any
+) -> timezone | None:
+ """Recover the offset selected by an exact dateutil named timezone.
+
+ A dateutil tzfile's finite transition table is its wire-semantic source of
+ truth. Reinterpreting its wall time through a system ``ZoneInfo`` database
+ changes negative-DST folds, nonexistent times, and dates after the final
+ transition. This mirrors dateutil's transition selection using only exact
+ builtin containers and its exact trusted transition-record type.
+ """
+ if _dateutil_timezone_name_without_hooks(tzinfo) is None:
+ return None
+ namespace = _exact_object_namespace(tzinfo)
+ if namespace is None:
+ return None
+ transitions = dict.get(namespace, "_trans_list")
+ transition_info = dict.get(namespace, "_trans_idx")
+ standard_info = dict.get(namespace, "_ttinfo_std")
+ before_info = dict.get(namespace, "_ttinfo_before")
+ transition_count = tuple.__len__(transitions) if type(transitions) is
tuple else 0
+ if (
+ type(transitions) is not tuple
+ or type(transition_info) is not tuple
+ or tuple.__len__(transitions) != tuple.__len__(transition_info)
+ or transition_count > _MAX_DATEUTIL_TRANSITIONS
+ or _dateutil_ttinfo_without_hooks(standard_info) is None
+ or (
+ transition_count > 0 and
_dateutil_ttinfo_without_hooks(before_info) is None
+ )
+ ):
+ return None
+
+ previous: int | None = None
+ for transition in transitions:
+ if (
+ type(transition) is not int
+ or int.bit_length(transition) > 63
+ or (previous is not None and transition < previous)
+ ):
+ return None
+ previous = transition
+ if any(_dateutil_ttinfo_without_hooks(info) is None for info in
transition_info):
+ return None
+
+ naive = datetime(
+ value.year,
+ value.month,
+ value.day,
+ value.hour,
+ value.minute,
+ value.second,
+ value.microsecond,
+ )
+ try:
+ timestamp = (naive - EPOCH).total_seconds()
+ except (OverflowError, TypeError, ValueError):
+ return None
+
+ index: int | None = (
+ bisect_right(transitions, timestamp) - 1 if transition_count else None
+ )
+
+ def info_at(selected: int | None) -> Any:
+ if selected is None or selected + 1 >= transition_count:
+ return standard_info
+ if selected < 0:
+ return before_info
+ return tuple.__getitem__(transition_info, selected)
+
+ if index is not None and index != 0:
+ current = _dateutil_ttinfo_without_hooks(info_at(index))
+ prior = _dateutil_ttinfo_without_hooks(info_at(index - 1))
+ if current is None or prior is None:
+ return None
+ offset_delta = prior[0] - current[0]
+ transition = tuple.__getitem__(transitions, index)
+ is_ambiguous = timestamp < transition + offset_delta
+ index -= int(not value.fold and is_ambiguous)
+
+ selected = _dateutil_ttinfo_without_hooks(info_at(index))
+ if selected is None:
+ return None
+ try:
+ return timezone(selected[1])
+ except ValueError:
+ return None
+
+
+def _pytz_timezone_name_without_hooks(tzinfo: Any) -> str | None:
+ """Read and verify one generated pytz named-zone implementation."""
+ value_type = type(tzinfo)
+ if not _mro_contains(_type_mro(value_type), _PYTZ_NAMED_BASE_TYPES):
+ return None
+ try:
+ namespace = type.__getattribute__(value_type, "__dict__")
+ except (AttributeError, TypeError):
return None
+ if type(namespace) is not MappingProxyType:
+ return None
+ zone = namespace.get("zone")
+ if type(zone) is not str or _bounded_utf8_length(zone, 256) is None:
+ return None
+ try:
+ canonical = pytz.timezone(zone)
+ except (KeyError, ValueError):
+ return None
+ # A user subclass can inherit pytz's base and spoof ``zone``. Only the
+ # concrete class generated and cached by pytz for that name is trusted.
+ return zone if type(canonical) is value_type else None
+
+
+def _fixed_offset_without_hooks(tzinfo: Any) -> timezone | None:
+ """Reconstruct trusted dateutil/pytz fixed offsets from exact storage."""
+ if type(tzinfo) not in {_DATEUTIL_TZOFFSET_TYPE, _PYTZ_FIXED_OFFSET_TYPE}:
+ return None
+ namespace = _exact_object_namespace(tzinfo)
+ if namespace is None:
+ return None
+ offset = dict.get(namespace, "_offset")
+ if type(offset) is not timedelta:
+ return None
+ try:
+ return timezone(offset)
+ except ValueError:
+ return None
+
+
+def _pytz_named_offset_without_hooks(tzinfo: Any) -> timezone | None:
+ """Return a localized pytz instance's stored offset without its hooks."""
+ if _pytz_timezone_name_without_hooks(tzinfo) is None:
+ return None
+ namespace = _exact_object_namespace(tzinfo)
+ if namespace is None:
+ return None
+ offset = dict.get(namespace, "_utcoffset")
+ if type(offset) is not timedelta:
+ return None
+ try:
+ return timezone(offset)
+ except ValueError:
+ return None
+
+
+def _dateutil_local_offset_without_hooks(
+ value: datetime, tzinfo: Any
+) -> timezone | None:
+ """Select an exact dateutil-local offset using builtin system time data."""
+ if type(tzinfo) is not _DATEUTIL_TZLOCAL_TYPE:
+ return None
+ namespace = _exact_object_namespace(tzinfo)
+ if namespace is None:
+ return None
+ standard_offset = dict.get(namespace, "_std_offset")
+ daylight_offset = dict.get(namespace, "_dst_offset")
+ has_daylight = dict.get(namespace, "_hasdst")
+ if (
+ type(standard_offset) is not timedelta
+ or type(daylight_offset) is not timedelta
+ or type(has_daylight) is not bool
+ ):
+ return None
+ selected_offset = standard_offset
+ if has_daylight:
+ epoch = datetime(1970, 1, 1)
+ naive = datetime(
+ value.year,
+ value.month,
+ value.day,
+ value.hour,
+ value.minute,
+ value.second,
+ value.microsecond,
+ )
+ timestamp = (naive - epoch).total_seconds()
+ try:
+ is_daylight = bool(
+ system_time.localtime(timestamp +
system_time.timezone).tm_isdst
+ )
+ daylight_saved = daylight_offset - standard_offset
+ previous_is_daylight = bool(
+ system_time.localtime(
+ timestamp
+ - timedelta.total_seconds(daylight_saved)
+ + system_time.timezone
+ ).tm_isdst
+ )
+ except (OverflowError, OSError, ValueError):
+ return None
+ is_ambiguous = not is_daylight and is_daylight != previous_is_daylight
+ if is_ambiguous:
+ is_daylight = not bool(value.fold)
+ selected_offset = daylight_offset if is_daylight else standard_offset
+ try:
+ return timezone(selected_offset)
+ except ValueError:
+ return None
+
+
+def _canonical_timezone(tzinfo: Any) -> timezone | ZoneInfo | None:
+ """Return an exact trusted timezone without invoking the source's
methods."""
+ if any(type(tzinfo) is type_ for type_ in _TRUSTED_TZINFO_TYPES):
+ return tzinfo
+ if type(tzinfo) in {_DATEUTIL_TZUTC_TYPE, _PYTZ_UTC_TYPE}:
+ return timezone.utc
+ if fixed_offset := _fixed_offset_without_hooks(tzinfo):
+ return fixed_offset
+ zone_name = _dateutil_timezone_name_without_hooks(
+ tzinfo
+ ) or _pytz_timezone_name_without_hooks(tzinfo)
+ if zone_name:
+ try:
+ return ZoneInfo(zone_name)
+ except (KeyError, ValueError, ZoneInfoNotFoundError):
+ return None
+ return None
+
+
+def _timestamp_offset_without_hooks(value: pd.Timestamp) -> timezone | None:
+ """Recover a timestamp's stored wall-clock offset without timezone
hooks."""
+ unit_multipliers = {"s": 1_000_000_000, "ms": 1_000_000, "us": 1_000,
"ns": 1}
+ multiplier = unit_multipliers.get(value.unit)
+ if multiplier is None:
+ return None
+ try:
+ instant_ns = int(value.asm8.view("i8")) * multiplier
+ epoch_ordinal = date.toordinal(date(1970, 1, 1))
+ wall_ns = (
+ (
+ (datetime.toordinal(value) - epoch_ordinal) * 86_400
+ + value.hour * 3600
+ + value.minute * 60
+ + value.second
+ )
+ * 1_000_000_000
+ + value.microsecond * 1000
+ + value.nanosecond
+ )
+ offset_ns = wall_ns - instant_ns
+ if offset_ns % 1000:
+ return None
+ return timezone(timedelta(microseconds=offset_ns // 1000))
+ except (OverflowError, TypeError, ValueError):
+ return None
+
+
+def _trusted_datetime_value(
+ value: datetime,
+) -> tuple[datetime | None, str | None]:
+ """Return an exact datetime rebuilt with only trusted timezone types."""
+ tzinfo = value.tzinfo
+ canonical_value = value
+ if tzinfo is not None and not any(
+ type(tzinfo) is trusted for trusted in _TRUSTED_TZINFO_TYPES
+ ):
+ canonical_tz: timezone | ZoneInfo | None
+ if _dateutil_timezone_name_without_hooks(tzinfo) is not None:
+ # A recognized dateutil tzfile must use its own finite transition
+ # table. Falling through to ZoneInfo would silently reinterpret a
+ # source-selected gap/fold or post-table wall time.
+ canonical_tz = _dateutil_named_offset_without_hooks(value, tzinfo)
+ else:
+ canonical_tz = (
+ _pytz_named_offset_without_hooks(tzinfo)
+ or _dateutil_local_offset_without_hooks(value, tzinfo)
+ or _canonical_timezone(tzinfo)
+ )
+ if canonical_tz is None:
+ return None, "contains a datetime with an unsupported timezone"
+ canonical_value = datetime(
+ value.year,
+ value.month,
+ value.day,
+ value.hour,
+ value.minute,
+ value.second,
+ value.microsecond,
+ tzinfo=canonical_tz,
+ fold=value.fold,
+ )
+ try:
+ # Exercise builtin validation without dispatching through a source
+ # timezone after the reconstruction above.
+ datetime.isoformat(canonical_value)
+ except (OverflowError, TypeError, ValueError):
+ return None, "contains an invalid datetime"
+ return canonical_value, None
+
+
+def _trusted_datetime_text(value: datetime) -> tuple[str | None, str | None]:
+ """Serialize an exact Python datetime through only trusted timezone
types."""
+ canonical_value, reason = _trusted_datetime_value(value)
+ if reason is not None or canonical_value is None:
+ return None, reason or "contains an invalid datetime"
+ return datetime.isoformat(canonical_value), None
+
+
+def _trusted_time_text(value: time) -> tuple[str | None, str | None]:
+ """Serialize an exact Python time through only trusted timezone types."""
+ tzinfo = value.tzinfo
+ canonical_value = value
+ if tzinfo is not None and not any(
+ type(tzinfo) is trusted for trusted in _TRUSTED_TZINFO_TYPES
+ ):
+ canonical_tz = _canonical_timezone(tzinfo)
+ if canonical_tz is None and type(tzinfo) is _DATEUTIL_TZLOCAL_TYPE:
+ namespace = _exact_object_namespace(tzinfo)
+ if namespace is None or type(dict.get(namespace, "_hasdst")) is
not bool:
+ return None, "contains a time with an unsupported timezone"
+ if dict.get(namespace, "_hasdst"):
+ canonical_tz = None
+ else:
+ standard_offset = dict.get(namespace, "_std_offset")
+ if type(standard_offset) is not timedelta:
+ return None, "contains a time with an unsupported timezone"
+ try:
+ canonical_tz = timezone(standard_offset)
+ except ValueError:
+ return None, "contains a time with an unsupported timezone"
+ elif canonical_tz is None:
+ return None, "contains a time with an unsupported timezone"
+ canonical_value = time(
+ value.hour,
+ value.minute,
+ value.second,
+ value.microsecond,
+ tzinfo=canonical_tz,
+ fold=value.fold,
+ )
+ try:
+ return time.isoformat(canonical_value), None
+ except (OverflowError, TypeError, ValueError):
+ return None, "contains an invalid time"
+
+
+def _trusted_timestamp_value(
+ value: pd.Timestamp,
+) -> tuple[pd.Timestamp | None, str | None]:
+ """Return a timestamp rebuilt with only trusted timezone
implementations."""
+ tzinfo = value.tzinfo
+ try:
+ if tzinfo is not None and not any(
+ type(tzinfo) is trusted for trusted in _TRUSTED_TZINFO_TYPES
+ ):
+ if (
+ _canonical_timezone(tzinfo) is None
+ and type(tzinfo) is not _DATEUTIL_TZLOCAL_TYPE
+ ):
+ return (
+ None,
+ "contains a pandas timestamp with an unsupported timezone",
+ )
+ if (canonical_tz := _timestamp_offset_without_hooks(value)) is
None:
+ return None, "contains an invalid pandas timestamp"
+ # Rebuild from the stored instant and resolution. No method on the
+ # original pytz/dateutil object is called, and the recovered fixed
+ # offset preserves the timestamp's selected fold.
+ raw_value = value.asm8.view("i8")
+ value = pd.Timestamp(raw_value, unit=value.unit,
tz="UTC").tz_convert(
+ canonical_tz
+ )
+ # Validate the retained resolution and selected UTC offset.
+ pd.Timestamp.isoformat(value)
+ except (KeyError, OverflowError, TypeError, ValueError):
+ return None, "contains an invalid pandas timestamp"
+ return value, None
+
+
+def _trusted_timestamp_text(value: pd.Timestamp) -> tuple[str | None, str |
None]:
+ """Convert an exact pandas timestamp to its canonical JSON
representation."""
+ canonical_value, reason = _trusted_timestamp_value(value)
+ if reason is not None or canonical_value is None:
+ return None, reason or "contains an invalid pandas timestamp"
+ # ISO output preserves nanoseconds and the UTC offset selected by fold.
+ text = pd.Timestamp.isoformat(canonical_value)
+ if _bounded_utf8_length(text, MAX_QUERY_RESULT_STRING_BYTES) is None:
+ return None, "contains an oversized pandas timestamp"
+ return text, None
+
+
+def _normalize_trusted_scalar( # noqa: C901
+ value: Any, *, max_string_bytes: int = MAX_QUERY_RESULT_STRING_BYTES
+) -> tuple[Any, str | None]:
+ """Normalize one exact trusted pandas/NumPy scalar or validate a builtin.
+
+ Type identity is checked before every conversion. This deliberately does
not
+ accept subclasses or generic ``np.generic``/pandas extension objects, whose
+ conversion hooks are outside the trusted ChartData materialization
contract.
+ """
+ value_type = type(value)
+ enum_seen: set[int] = set()
+ while _mro_contains(_type_mro(value_type), (Enum,)):
+ identity = id(value)
+ if identity in enum_seen or len(enum_seen) >= _MAX_ROW_CONTAINER_DEPTH:
+ return None, "contains a recursive enum"
+ enum_seen.add(identity)
+ try:
+ value = object.__getattribute__(value, "_value_")
+ except Exception:
+ return None, "contains an unsupported enum"
+ value_type = type(value)
+
+ if value is None or value_type is bool:
+ return value, None
+ if value_type is str:
+ size = _bounded_utf8_length(value, max_string_bytes)
+ return (
+ (value, None)
+ if size is not None
+ else (
+ None,
+ "contains an invalid or oversized string",
+ )
+ )
+ if value_type is int:
+ return value, _integer_failure(value)
+ if value_type is float:
+ if math.isnan(value):
+ return None, None
+ if math.isinf(value):
+ return None, "contains a non-finite number"
Review Comment:
`df_to_records` nulls top-level infinities, but an infinity inside an array
cell (say a Postgres `float8[]` holding `'Infinity'`) reaches this branch and
fails the entire `query_dataset`/`get_table` result with
`MalformedQueryResult`. The NaN case a few lines up already maps to `None`, so
should a nested infinity be nulled the same way instead of rejecting every row?
##########
superset/mcp_service/chart/chart_utils.py:
##########
@@ -1715,6 +1758,853 @@ def map_histogram_config(config:
"HistogramChartConfig") -> Dict[str, Any]:
return form_data
+def _bullet_token_list(values: Sequence[str | int | float]) -> str:
+ """Serialize typed Bullet controls to the frontend's comma-separated
form."""
+ tokens: list[str] = []
+ for value in values:
+ if isinstance(value, float):
+ token = repr(value)
+ # ``100`` parses back to the same binary float as ``100.0`` and
+ # preserves the frontend's established compact integer spelling.
+ if token.endswith(".0") and not (
+ value == 0.0 and math.copysign(1.0, value) < 0
+ ):
+ token = token[:-2]
+ tokens.append(token)
+ else:
+ tokens.append(str(value))
+ return ",".join(tokens)
+
+
+def map_bullet_config(config: BulletChartConfig) -> Dict[str, Any]: # noqa:
C901
+ """Map typed Bullet config to ``Bullet/buildQuery`` and transformProps.
+
+ The frontend buildQuery replaces the generic query fields with exactly one
+ metric and the groupby hierarchy. Presentation controls stay in native
+ snake_case form_data; the chart plugin camelizes them for transformProps.
+ """
+ if (
+ config.dimensions is None
+ and config._inherited_groupby is None
+ and config.order_by
+ ):
+ # An update resolves its saved hierarchy before mapping. Without one,
+ # creation must validate sort targets against an empty hierarchy.
+ BulletChartConfig.model_validate(
+ {**config.model_dump(exclude_unset=True), "dimensions": []}
+ )
+ metric = create_metric_object(config.metric)
+ form_data: Dict[str, Any] = {
+ "viz_type": "bullet",
+ "metric": metric,
+ }
+
+ # Optional semantic/query fields are emitted only when explicitly supplied.
+ # This lets update_chart and update_chart_preview preserve native saved
state,
+ # while an explicit empty value still clears it through the generic merge
path.
+ # The row limit always carries the schema default so a new chart is
bounded;
+ # update merging restores the saved limit when the caller omits it.
+ if "dimensions" in config.model_fields_set:
+ form_data["groupby"] = [dimension.name for dimension in
config.dimensions or []]
+ form_data["row_limit"] = config.row_limit
+ if "time_range" in config.model_fields_set:
+ form_data["time_range"] = config.time_range
+
+ if config.order_by:
+ dimensions = config.order_dimensions
+ orderby: list[list[Any]] = []
+ for order in config.order_by:
+ role, index = resolve_bullet_order_target(
+ order.column, dimensions, config.metric
+ )
+ if role == "metric":
+ order_target: Any = metric
+ else:
+ if index is None: # Defensive: resolver pairs dimensions with
indexes.
+ raise ValueError("Bullet dimension order target has no
index")
+ dimension = dimensions[index]
+ order_target = (
+ dimension.name if isinstance(dimension, ColumnRef) else
dimension
+ )
+ orderby.append([order_target, order.ascending])
+ form_data["orderby"] = orderby
+ elif "order_by" in config.model_fields_set:
+ form_data["orderby"] = []
+
+ presentation_fields: dict[str, tuple[str, Any]] = {
+ "ranges": ("ranges", _bullet_token_list(config.ranges)),
+ "range_labels": (
+ "range_labels",
+ _bullet_token_list(config.range_labels),
+ ),
+ "markers": ("markers", _bullet_token_list(config.markers)),
+ "marker_labels": (
+ "marker_labels",
+ _bullet_token_list(config.marker_labels),
+ ),
+ "marker_lines": (
+ "marker_lines",
+ _bullet_token_list(config.marker_lines),
+ ),
+ "marker_line_labels": (
+ "marker_line_labels",
+ _bullet_token_list(config.marker_line_labels),
+ ),
+ "y_axis_format": ("y_axis_format", config.y_axis_format),
+ "show_labels": ("show_labels", config.show_labels),
+ "show_legend": ("show_legend", config.show_legend),
+ }
+ for field_name, (form_key, value) in presentation_fields.items():
+ if field_name in config.model_fields_set:
+ form_data[form_key] = value
+
+ _add_adhoc_filters(form_data, config.filters)
+ if config.filters == [] and "filters" in config.model_fields_set:
+ form_data["adhoc_filters"] = []
+ if config.time_range and config.temporal_column:
+ _ensure_temporal_adhoc_filter(form_data, config.temporal_column)
+ for filter_ in form_data.get("adhoc_filters", []):
+ if (
+ isinstance(filter_, dict)
+ and filter_.get("operator") ==
FilterOperator.TEMPORAL_RANGE.value
+ and filter_.get("subject") == config.temporal_column
+ and filter_.get("comparator") == NO_TIME_RANGE
+ ):
+ filter_["comparator"] = config.time_range
+ return form_data
+
+
+def _normalize_native_filter_aliases(form_data: Mapping[str, Any]) ->
Dict[str, Any]:
+ """Fold legacy WHERE, HAVING, and filter predicates into adhoc controls."""
+ from superset.utils.core import form_data_to_adhoc, simple_filter_to_adhoc
+
+ normalized = dict(form_data)
+ legacy_filters = [
+ form_data_to_adhoc(normalized, clause)
+ for clause in ("having", "where")
+ if normalized.get(clause)
+ ]
+ legacy_filters.extend(
+ simple_filter_to_adhoc(filter_, "where")
+ for filter_ in normalized.get("filters") or []
+ if filter_ is not None
+ )
+ if legacy_filters:
+ normalized["adhoc_filters"] = [
+ *legacy_filters,
+ *(normalized.get("adhoc_filters") or []),
+ ]
+ for key in ("where", "having", "filters"):
+ normalized.pop(key, None)
+ return normalized
+
+
+def _normalize_bullet_query_aliases(form_data: Mapping[str, Any]) -> Dict[str,
Any]:
+ """Fold inherited native predicates and ordering into canonical
controls."""
+ from superset.mcp_service.chart.chart_helpers import _parse_orderby
+
+ normalized = _normalize_native_filter_aliases(form_data)
+ if "groupby" in normalized and not isinstance(normalized["groupby"], list):
+ # Bullet/buildQuery and transformProps read ensureIsArray(groupby).
+ normalized["groupby"] = bullet_groupby_list(normalized["groupby"])
+ if "order_by_cols" in normalized:
+ # Native extractQueryFields concatenates both aliases in key order.
+ ordering: list[Any] = []
+ for key, value in normalized.items():
+ if key == "order_by_cols":
+ ordering.extend(_parse_orderby(value))
+ elif key == "orderby":
+ ordering.extend(value or [])
+ normalized["orderby"] = ordering
+ normalized.pop("order_by_cols", None)
+ return normalized
+
+
+def merge_bullet_form_data(
+ existing_form_data: Mapping[str, Any], new_form_data: Dict[str, Any]
+) -> None:
+ """Preserve omitted native Bullet controls across update tool paths.
+
+ Query roles and every UI control have an explicit typed representation.
+ Mappers emit optional fields only when the caller supplied them, so copying
+ the bounded native keys below preserves omitted state while explicit empty,
+ false, null, and zero-like values remain authoritative.
+ """
+ if (
+ existing_form_data.get("viz_type") != "bullet"
+ or new_form_data.get("viz_type") != "bullet"
+ ):
+ return
+ existing_form_data = _normalize_bullet_query_aliases(existing_form_data)
+ preserved_keys = {
+ "groupby",
+ "adhoc_filters",
+ "time_range",
+ "row_limit",
+ "orderby",
+ "ranges",
+ "range_labels",
+ "markers",
+ "marker_labels",
+ "marker_lines",
+ "marker_line_labels",
+ "y_axis_format",
+ "show_labels",
+ "show_legend",
+ "url_params",
+ # Native query context (dashboard/native filter predicates and time
+ # overrides) that buildQueryObject applies on top of the controls.
+ "extra_form_data",
+ "extra_filters",
+ MCP_DASHBOARD_TIME_FILTER_SUBJECT,
+ }
+
+ # Threshold and label arrays are one frontend control pair. If callers
+ # replace the values without replacing their labels, clear the stale labels
+ # instead of accidentally reassigning them by position.
+ dependent_controls = {
+ "ranges": "range_labels",
+ "markers": "marker_labels",
+ "marker_lines": "marker_line_labels",
+ }
+ for values_key, labels_key in dependent_controls.items():
+ if values_key in new_form_data and labels_key not in new_form_data:
+ new_form_data[labels_key] = ""
+
+ preserve_orderby = (
+ "orderby" not in new_form_data and "orderby" in existing_form_data
+ )
+ for key in preserved_keys:
+ if (
+ key == MCP_DASHBOARD_TIME_FILTER_SUBJECT
+ and "adhoc_filters" in new_form_data
+ ):
+ # The marker describes a mapper-generated temporal filter. Do not
+ # retain stale provenance when an explicit filter update removed
it.
+ continue
+ if key in existing_form_data and key not in new_form_data:
+ new_form_data[key] = existing_form_data[key]
+ if preserve_orderby:
+ new_form_data["orderby"] = _orderby_for_final_output_roles(
+ existing_form_data, new_form_data
+ )
+
+
+def _bullet_output_labels(
+ form_data: Mapping[str, Any],
+) -> tuple[set[str], dict[str, Any]]:
+ """Return a Bullet state's dimension output labels and metric outputs."""
+ from superset.mcp_service.chart.chart_helpers import _column_label,
_metric_label
+
+ dimensions = {
+ label
+ for column in bullet_groupby_list(form_data.get("groupby"))
+ if (label := _column_label(column)) is not None
+ }
+ metrics = form_data.get("metrics") or []
+ if not isinstance(metrics, (list, tuple)):
+ metrics = [metrics]
+ metric_outputs = {
+ label: metric
+ for metric in [form_data.get("metric"), *metrics]
+ if (label := _metric_label(metric)) is not None
+ }
+ return dimensions, metric_outputs
+
+
+def bullet_groupby_list(groupby: Any) -> list[Any]:
+ """Normalize a saved Bullet hierarchy like the frontend
``ensureIsArray``."""
+ if groupby is None:
+ return []
+ return list(groupby) if isinstance(groupby, (list, tuple)) else [groupby]
+
+
+def _orderby_for_final_output_roles(
+ existing_form_data: Mapping[str, Any], new_form_data: Mapping[str, Any]
+) -> Any:
+ """Drop sorts on removed output roles and rebind inherited metric
expressions.
+
+ Native ordering may also rank by a saved metric or column that is not a
+ displayed output (``get_sqla_query`` resolves it independently). Those
+ sorters never named a Bullet role, so a role change does not remove them.
+ """
+ from superset.mcp_service.chart.chart_helpers import _column_label,
_metric_label
+
+ saved = existing_form_data.get("orderby")
+ if not isinstance(saved, list):
+ return saved
+ outputs, metric_outputs = _bullet_output_labels(new_form_data)
+ outputs.update(metric_outputs)
+ previous_dimensions, previous_metrics =
_bullet_output_labels(existing_form_data)
+ previous_outputs = previous_dimensions | set(previous_metrics)
+ retained = []
+ for entry in saved:
+ if isinstance(entry, (list, tuple)) and entry:
+ target = entry[0]
+ label = (
+ _metric_label(target)
+ or _column_label(target)
+ or target.get("metric_name")
+ if isinstance(target, Mapping)
+ else target
+ )
+ if (
+ isinstance(label, str)
+ and label not in outputs
+ and label in previous_outputs
+ ):
+ continue
+ if (
+ isinstance(target, Mapping)
+ and isinstance(label, str)
+ and label in metric_outputs
+ ):
+ # Label equality identifies an output role, not expression
+ # equality: execute the final metric, never the saved
expression.
+ entry = [metric_outputs[label], *entry[1:]]
+ retained.append(entry)
+ return retained
+
+
+def _filter_identity(filter_: Any) -> tuple[Any, ...] | None:
+ """Return the native identity used when one filter replaces another."""
+ if not isinstance(filter_, Mapping):
+ return None
+ return (
+ filter_.get("clause"),
+ filter_.get("expressionType"),
+ filter_.get("subject"),
+ filter_.get("operator"),
+ )
+
+
+def _temporal_binding_filter(filters: list[Any], subject: Any) -> dict[str,
Any] | None:
+ """Find the unique filter owned by a recorded MCP temporal marker."""
+ if subject is None:
+ return None
+ if not isinstance(subject, str) or not subject:
+ raise ValueError(
+ "MCP temporal binding provenance subject must be a non-empty
string"
+ )
+ matches = [
+ filter_
+ for filter_ in filters
+ if isinstance(filter_, dict)
+ and filter_.get("subject") == subject
+ and filter_.get("operator") == FilterOperator.TEMPORAL_RANGE.value
+ ]
+ if len(matches) != 1:
+ raise ValueError(
+ "MCP temporal binding provenance must match exactly one "
+ f"TEMPORAL_RANGE filter for subject {subject!r}; found
{len(matches)}"
+ )
+ return matches[0]
+
+
+def _append_or_replace_filter(filters: list[Any], filter_: Any) -> None:
+ """Append a filter, replacing the same native role when identifiable."""
+ identity = _filter_identity(filter_)
+ if identity is None:
+ if filter_ not in filters:
+ filters.append(filter_)
+ return
+ filters[:] = [item for item in filters if _filter_identity(item) !=
identity]
+ filters.append(filter_)
+
+
+_NATIVE_TEMPORAL_ROLE_FIELDS: dict[str, frozenset[str]] = {
+ # Typed ``x`` is persisted as native x_axis/granularity_sqla for XY and
+ # Mixed Timeseries. Waterfall exposes the typed field as ``x_axis``.
+ "x_axis": frozenset({"x", "x_axis"}),
+ "granularity_sqla": frozenset({"x", "x_axis", "temporal_column"}),
+ # Chart plugins may designate a chart-specific query role as the implicit
+ # dashboard-time subject.
+ "start_time": frozenset({"start_time"}),
+}
+
+
+def _native_temporal_subject_changed(
+ existing_form_data: Mapping[str, Any],
+ new_form_data: Mapping[str, Any],
+ explicit_fields: set[str],
+) -> bool:
+ """Return whether an authoritative native temporal role was replaced.
+
+ Mapping a partial update can propose a dataset fallback binding even when
+ the caller only changed filters. That proposal is not authoritative. A
+ changed x/granularity/chart-specific role is authoritative only when its
+ corresponding typed field was actually supplied.
+ """
+ for native_key, typed_fields in _NATIVE_TEMPORAL_ROLE_FIELDS.items():
+ if explicit_fields.isdisjoint(typed_fields):
+ continue
+ existing_value = existing_form_data.get(native_key)
+ incoming_value = new_form_data.get(native_key)
+ if existing_value != incoming_value:
+ return True
+ return False
+
+
+def _native_temporal_binding(
+ form_data: Mapping[str, Any], filters: list[Any]
+) -> tuple[str | None, dict[str, Any] | None]:
+ """Resolve one binding for a trusted native temporal role, if present."""
+ for native_key in _NATIVE_TEMPORAL_ROLE_FIELDS:
+ subject = form_data.get(native_key)
+ if not isinstance(subject, str) or not subject:
+ continue
+ matches = [
+ filter_
+ for filter_ in filters
+ if isinstance(filter_, dict)
+ and filter_.get("subject") == subject
+ and filter_.get("operator") == FilterOperator.TEMPORAL_RANGE.value
+ ]
+ if len(matches) > 1:
+ raise ValueError(
+ "An authoritative native temporal subject must match at most
one "
+ f"TEMPORAL_RANGE filter for subject {subject!r}; found "
+ f"{len(matches)}"
+ )
+ if matches:
+ return subject, matches[0]
+ return None, None
+
+
+def merge_update_form_data( # noqa: C901
+ existing_form_data: Mapping[str, Any],
+ new_form_data: Dict[str, Any],
+ config: ChartConfig,
+) -> None:
+ """Apply the shared omission/provenance contract for chart updates.
+
+ Mapper-generated neutral temporal bindings are infrastructure, not evidence
+ that the caller supplied ``filters`` or changed a saved time-range binding.
+ This helper is used by immediate saves, preview-first saved updates, and
+ cached-preview updates so omission, clear, replacement, and temporal
+ overrides have identical behavior.
+
+ State never crosses a visualization boundary: a viz-type change starts from
+ the mapper's output, so the previous chart's predicates are not restored.
+ """
+ existing_viz_type = existing_form_data.get("viz_type")
+ if isinstance(existing_viz_type, str) and existing_viz_type !=
new_form_data.get(
+ "viz_type"
+ ):
+ return
+ existing_form_data = _normalize_native_filter_aliases(existing_form_data)
+ # The initial overlay may carry legacy keys from saved form data. Filter
+ # omission/replacement below owns the complete canonical predicate
sequence.
+ for key in ("where", "having", "filters"):
+ new_form_data.pop(key, None)
+ existing_filters = list(existing_form_data.get("adhoc_filters") or [])
+ incoming_filters = list(new_form_data.get("adhoc_filters") or [])
+ existing_subject =
existing_form_data.get(MCP_DASHBOARD_TIME_FILTER_SUBJECT)
+ incoming_subject = new_form_data.get(MCP_DASHBOARD_TIME_FILTER_SUBJECT)
+ existing_binding = _temporal_binding_filter(existing_filters,
existing_subject)
+ incoming_binding = _temporal_binding_filter(incoming_filters,
incoming_subject)
+
+ explicit_fields = set(getattr(config, "model_fields_set", set()))
+ filters_explicit = "filters" in explicit_fields
+ range_explicit = "time_range" in explicit_fields
+ subject_explicit = "temporal_column" in explicit_fields
+ native_subject_changed = _native_temporal_subject_changed(
+ existing_form_data, new_form_data, explicit_fields
+ )
+ subject_authoritative = subject_explicit or native_subject_changed
+ if incoming_binding is None:
+ native_subject, native_binding = _native_temporal_binding(
+ new_form_data, incoming_filters
+ )
+ if native_binding is not None:
+ incoming_subject = native_subject
+ incoming_binding = native_binding
+ incoming_user_filters = [
+ filter_ for filter_ in incoming_filters if filter_ is not
incoming_binding
+ ]
+ temporal_explicit = range_explicit or subject_authoritative
+ if (
+ existing_binding is None
+ and incoming_binding is not None
+ and isinstance(incoming_subject, str)
+ and "filters" not in explicit_fields
+ and temporal_explicit
+ ):
+ # A saved temporal filter that Explore wrote has no MCP provenance
+ # marker. When it is the only native filter for the incoming subject,
+ # the update replaces it in place instead of appending a duplicate.
+ native_matches = [
+ filter_
+ for filter_ in existing_filters
+ if isinstance(filter_, dict)
+ and filter_.get("subject") == incoming_subject
+ and filter_.get("operator") == FilterOperator.TEMPORAL_RANGE.value
+ ]
+ if len(native_matches) == 1:
+ existing_binding = native_matches[0]
+ existing_subject = incoming_subject
+
+ chosen_binding: dict[str, Any] | None = None
+ chosen_subject: Any = None
+ if not filters_explicit:
+ # Omission is byte-faithful: keep the native sequence in its exact
order,
+ # including SQL/HAVING objects and a provenance-owned binding at any
index.
+ merged_filters = list(existing_filters)
+ chosen_binding = existing_binding
+ chosen_subject = existing_subject
+ if temporal_explicit:
+ if subject_authoritative:
+ chosen_binding = incoming_binding
+ chosen_subject = incoming_subject
+ elif existing_binding is not None:
+ # A range-only update belongs to the saved subject, even when
+ # mapping the partial config proposed the dataset main_dttm.
+ chosen_binding = dict(existing_binding)
+ chosen_subject = existing_subject
+ else:
+ chosen_binding = incoming_binding
+ chosen_subject = incoming_subject
+
+ if chosen_binding is not None:
+ chosen_binding = dict(chosen_binding)
+ if range_explicit:
+ chosen_binding["comparator"] = (
+ getattr(config, "time_range", None) or NO_TIME_RANGE
+ )
+ elif existing_binding is not None:
+ # Subject-only replacement preserves the saved active or
+ # neutral range instead of resetting it to No filter.
+ chosen_binding["comparator"] = existing_binding.get(
+ "comparator", NO_TIME_RANGE
+ )
+ if existing_binding is not None:
+ binding_index = next(
+ index
+ for index, filter_ in enumerate(merged_filters)
+ if filter_ is existing_binding
+ )
+ if chosen_binding is None:
+ merged_filters.pop(binding_index)
+ else:
+ # A temporal override changes infrastructure in place
instead
+ # of moving it past surrounding native filters.
+ merged_filters[binding_index] = chosen_binding
+ elif chosen_binding is not None:
+ merged_filters.append(chosen_binding)
+ else:
+ # An explicit filter array replaces the saved native sequence. The
mapper
+ # deliberately emits [] for an explicit clear; otherwise retain its
+ # generated temporal binding after the replacement filters.
+ merged_filters = list(incoming_user_filters)
+ if incoming_user_filters or temporal_explicit:
+ if subject_authoritative:
+ chosen_binding = incoming_binding
+ chosen_subject = incoming_subject
+ elif range_explicit and existing_binding is not None:
+ chosen_binding = dict(existing_binding)
+ chosen_subject = existing_subject
+ else:
+ # A filter-only replacement keeps the saved provenance binding.
+ # The mapper's incoming binding may merely be a dataset
fallback
+ # and must not reset the saved subject or active range.
+ chosen_binding = existing_binding
+ chosen_subject = existing_subject
+ if chosen_binding is not None:
+ chosen_binding = dict(chosen_binding)
+ if range_explicit:
+ chosen_binding["comparator"] = (
+ getattr(config, "time_range", None) or NO_TIME_RANGE
+ )
+ elif subject_authoritative and existing_binding is not None:
+ chosen_binding["comparator"] = existing_binding.get(
+ "comparator", NO_TIME_RANGE
+ )
+ if chosen_binding is not None:
+ _append_or_replace_filter(merged_filters, chosen_binding)
+
+ # Materialize exactly when saved state had the key or the caller made the
+ # controls authoritative. An omitted update must not turn a missing native
+ # filter key into [] merely because its mapper proposed a neutral binding.
+ if filters_explicit or "adhoc_filters" in existing_form_data or
temporal_explicit:
+ new_form_data["adhoc_filters"] = merged_filters
+ else:
+ new_form_data.pop("adhoc_filters", None)
+ if chosen_binding is not None and isinstance(chosen_subject, str):
+ new_form_data[MCP_DASHBOARD_TIME_FILTER_SUBJECT] = chosen_subject
+ else:
+ new_form_data.pop(MCP_DASHBOARD_TIME_FILTER_SUBJECT, None)
+
+
+def _currency_form_value(value: CurrencyFormat | None) -> dict[str, str] |
None:
+ """Return the native value for an explicitly supplied currency control."""
+ return value.to_form_data() if value is not None else None
+
+
+def _column_names(value: Sequence[ColumnRef] | None) -> list[str | None] |
None:
+ """Return a native column-name list while retaining an explicit null."""
+ return [column.name for column in value] if value is not None else None
+
+
+def _table_sort_value(value: Sequence[str | SortByConfig] | None) -> list[str]
| None:
+ """Return the native Table sort control for an explicit typed value."""
+ if value is None:
+ return None
+ return [
+ json.dumps(
+ [entry.column, entry.ascending]
+ if isinstance(entry, SortByConfig)
+ else [entry, False]
+ )
+ for entry in value
+ ]
+
+
+def _table_column_config_value(value: Any) -> dict[str, Any] | None:
+ """Return Table column config without losing an explicit null or empty
map."""
+ if value is None:
+ return None
+ return {
+ label: column.model_dump(by_alias=True, exclude_unset=True)
+ for label, column in value.items()
+ }
+
+
+# Mappers intentionally omit optional controls so fresh charts use the frontend
+# defaults. During a same-viz update, however, an explicitly supplied false,
+# null, or empty value must block preservation of the saved native key. Keep
the
+# typed-to-native relationship declarative so every update path shares it.
+_FormValueConverter = Callable[[Any], Any]
+_FormControlMap = dict[str, tuple[str, _FormValueConverter]]
+
+_COMMON_EXPLICIT_FORM_CONTROLS: _FormControlMap = {
+ "color_scheme": ("color_scheme", lambda value: value),
+ "currency_format": ("currency_format", _currency_form_value),
+ "show_value": ("show_value", lambda value: value),
+}
+
+_CHART_EXPLICIT_FORM_CONTROLS: dict[str, _FormControlMap] = {
+ "table": {
+ "sort_by": ("order_by_cols", _table_sort_value),
+ "column_config": ("column_config", _table_column_config_value),
+ },
+ "xy": {
+ "group_by": ("groupby", _column_names),
+ "series_limit": ("series_limit", lambda value: value),
+ "stacked": ("stack", lambda value: "Stack" if value else None),
+ "orientation": ("orientation", lambda value: value),
+ "legend_orientation": ("legendOrientation", lambda value: value),
+ "x_axis_time_format": ("x_axis_time_format", lambda value: value),
+ "time_grain": ("time_grain_sqla", lambda value: value),
+ },
+ "mixed_timeseries": {
+ "group_by": ("groupby", _column_names),
+ "group_by_secondary": ("groupby_b", _column_names),
+ "currency_format_secondary": (
+ "currency_format_secondary",
+ _currency_form_value,
+ ),
+ "time_grain": ("time_grain_sqla", lambda value: value),
+ },
+ "waterfall": {
+ "time_grain": ("time_grain_sqla", lambda value: value),
+ },
+ "big_number": {
+ "subheader": ("subheader", lambda value: value),
+ "y_axis_format": ("y_axis_format", lambda value: value),
+ "time_grain": ("time_grain_sqla", lambda value: value),
+ "compare_lag": ("compare_lag", lambda value: value),
+ "time_format": ("time_format", lambda value: value),
+ "aggregation": ("aggregation", lambda value: value),
+ },
+ "handlebars": {
+ "style_template": ("styleTemplate", lambda value: value),
+ "columns": ("all_columns", _column_names),
+ "groupby": ("groupby", _column_names),
+ "metrics": ("metrics", _column_names),
+ },
+ "pivot_table": {
+ "date_format": ("date_format", lambda value: value),
+ },
+ "interactive_pivot": {
+ "time_grain": ("time_grain_sqla", lambda value: value),
+ "series_limit": ("series_limit", lambda value: value),
+ "date_format": ("date_format", lambda value: value),
+ "column_sort": ("colOrder", lambda value: value),
+ },
+}
+
+
+def _apply_explicit_form_controls( # noqa: C901
+ existing_form_data: Mapping[str, Any],
+ new_form_data: Dict[str, Any],
+ config: ChartConfig,
+) -> None:
+ """Apply typed controls whose mapper omission represents a native clear."""
+ explicit_fields = set(getattr(config, "model_fields_set", set()))
+ controls = {
+ **_COMMON_EXPLICIT_FORM_CONTROLS,
+ **_CHART_EXPLICIT_FORM_CONTROLS.get(config.chart_type, {}),
+ }
+ for field_name, (native_key, convert) in controls.items():
+ if field_name in explicit_fields:
+ converted = convert(getattr(config, field_name))
+ is_clear = (
+ converted is None
+ or converted is False
+ or converted == ""
+ or converted in ([], {})
+ )
+ if is_clear:
+ new_form_data[native_key] = converted
+
+ axis_controls = {
+ "xy": (
+ ("x_axis", "x_axis_title", "x_axis_format", None),
+ ("y_axis", "y_axis_title", "y_axis_format", "logAxis"),
+ ),
+ "mixed_timeseries": (
+ ("x_axis", "xAxisTitle", "x_axis_time_format", None),
+ ("y_axis", "yAxisTitle", "y_axis_format", "logAxis"),
+ (
+ "y_axis_secondary",
+ "yAxisTitleSecondary",
+ "y_axis_format_secondary",
+ "logAxisSecondary",
+ ),
+ ),
+ }
+ if config.chart_type in axis_controls:
+ for field_name, title_key, format_key, scale_key in axis_controls[
+ config.chart_type
+ ]:
+ if field_name not in explicit_fields:
+ continue
+ axis = getattr(config, field_name)
+ if axis is None:
+ new_form_data[title_key] = None
+ new_form_data[format_key] = None
+ if scale_key:
+ new_form_data[scale_key] = None
+ continue
+ axis_fields = set(axis.model_fields_set)
+ if "title" in axis_fields:
+ new_form_data[title_key] = axis.title
+ if "format" in axis_fields:
+ new_form_data[format_key] = axis.format
+ if scale_key and "scale" in axis_fields:
+ new_form_data[scale_key] = (
+ None if axis.scale is None else axis.scale == "log"
+ )
+
+ if config.chart_type == "xy" and "legend" in explicit_fields:
+ legend = config.legend
+ if legend is None:
+ new_form_data["show_legend"] = None
+ new_form_data["legendOrientation"] = None
+ else:
+ legend_fields = set(legend.model_fields_set)
+ if "show" in legend_fields:
+ new_form_data["show_legend"] = legend.show
+ if "position" in legend_fields:
+ new_form_data["legendOrientation"] = legend.position
+
+ if config.chart_type == "interactive_pivot":
+ if "temporal_column" in explicit_fields and config.temporal_column is
None:
+ new_form_data["granularity_sqla"] = None
+ new_form_data["temporal_columns_lookup"] = None
+ if (
+ "series_limit_metric" in explicit_fields
+ and config.series_limit_metric is None
+ ):
+ new_form_data["series_limit_metric"] = None
+ if "comparison_period" in explicit_fields and config.comparison_period
is None:
+ new_form_data["time_compare"] = None
+ if "comparison_type" in explicit_fields and config.comparison_type is
None:
+ new_form_data["comparison_type"] = None
+
+ # A Waterfall axis replacement cannot inherit a bucket belonging to the old
+ # temporal subject. Grain omission preserves only while the axis is stable;
+ # explicit null is already handled by the declarative control map above.
+ if (
+ config.chart_type == "waterfall"
+ and existing_form_data.get("x_axis") != new_form_data.get("x_axis")
+ and "time_grain" not in explicit_fields
+ ):
+ new_form_data["time_grain_sqla"] = None
+
+
+def merge_same_viz_form_data(
+ existing_form_data: Mapping[str, Any],
+ new_form_data: Dict[str, Any],
+ config: ChartConfig,
+) -> None:
+ """Preserve saved controls that the typed mapper does not represent.
+
+ The typed MCP surface deliberately models a bounded subset of every Explore
+ control panel. For a replacement within the exact same native ``viz_type``,
+ keys absent from the mapper therefore represent omitted controls and retain
+ their saved values. Mapper output and the chart-specific merge helpers run
+ first and remain authoritative, including explicit empty, false, null, and
+ nested values.
+
+ No generic state crosses a visualization boundary. This prevents query-role
+ keys from the previous plugin (for example ``metric`` or ``groupby``) from
+ leaking into a different plugin whose role contract is unrelated.
+ """
+ existing_viz_type = existing_form_data.get("viz_type")
+ if not isinstance(existing_viz_type, str) or existing_viz_type !=
new_form_data.get(
+ "viz_type"
+ ):
+ return
+
+ _apply_explicit_form_controls(existing_form_data, new_form_data, config)
+
+ for key, value in existing_form_data.items():
+ if key in {"where", "having", "filters"}:
+ # merge_update_form_data already resolved these legacy predicates
+ # into adhoc_filters; restoring them would undo replacement/clear.
+ continue
+ if key == MCP_DASHBOARD_TIME_FILTER_SUBJECT:
+ # merge_update_form_data owns this provenance marker. Its absence
+ # may be an intentional subject clear and must not be undone by the
+ # generic preservation layer.
+ continue
+ if key not in new_form_data:
+ new_form_data[key] = value
+
+
+def validate_merged_bullet_form_data(
+ form_data: Mapping[str, Any],
+ update_config: ChartConfig | None = None,
+) -> BulletChartConfig | None:
+ """Validate final Bullet controls without reinterpreting inherited query
roles.
+
+ Saved Explore state may contain SQL dimensions and SQL WHERE/SIMPLE HAVING
+ filters beyond the typed authoring surface. Omitted roles are validated by
+ the native query contract and compilation, not as newly authored physical
+ columns or SIMPLE WHERE filters. Only this validation copy excludes them
+ and native query metadata; compiled and persisted form data stays intact.
+ Explicit replacements, including ``[]``, retain strict typed validation.
+ """
+ if form_data.get("viz_type") != "bullet":
+ return None
+ validation_data = dict(form_data)
+ for native_query_key in ("url_params", "extra_form_data", "extra_filters"):
+ validation_data.pop(native_query_key, None)
+ if update_config is None or isinstance(update_config, BulletChartConfig):
+ if update_config is None or update_config.dimensions is None:
+ validation_data.pop("groupby", None)
+ if update_config is None or "filters" not in
update_config.model_fields_set:
Review Comment:
On a fresh `update_chart_preview` (no `form_data_key`) with
`temporal_column: null` and any `filters` value, the mapper writes
`_mcp_dashboard_time_filter_subject: None`, there is no previous state merge to
drop it, and this branch keeps the marker once `filters` is set.
`BulletChartConfig` then raises "must be a physical column name" and the call
returns `invalid_merged_bullet_state` instead of a preview. Should the
validation copy drop a null marker regardless of `filters`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
