rebenitez1802 commented on code in PR #45348:
URL: https://github.com/apache/superset/pull/45348#discussion_r4289535263
##########
superset/sql/execution/celery_task.py:
##########
@@ -207,6 +209,13 @@ def _finalize_successful_query(
"execution_time_ms": exec_time,
}
)
+ if result_set.nested_columns:
+ # Keep supported nested values out of the outer
JSON/MessagePack
+ # serializer, which coerces types or rejects oversized
integers.
+ statements_data[-1]["nested_columns_msgpack"] = (
+ serialize_nested_columns(result_set.nested_columns)
+ )
Review Comment:
**🟡 Medium — Nested sidecar doubles nested payloads and counts in full
against the size limit**
This stores every nested column a second time, next to the Arrow data that
already holds it as JSON text. The copy is Base64 MessagePack, even in
MessagePack mode, and `_get_payload_size` (`:252-261`) counts it in full.
Repro: 50k JSONB-like rows with `SQLLAB_PAYLOAD_MAX_MB = 10`.
| Transport | master: stored size | this head: counted size and outcome |
| --- | --- | --- |
| MessagePack | 5.62 MiB | 10.31 MiB, `RESULT_TOO_LARGE` |
| JSON | 6.05 MiB | 10.31 MiB, `RESULT_TOO_LARGE` |
Even without a limit, stored bytes after zlib are about 1.9× master in
MessagePack mode and 2.7× in JSON mode. This is the same kind of issue as the
Base64 overhead that was just fixed, and it contradicts `mcp.md:50-51`.
Fix: store each nested column once (e.g. null out the Arrow column a sidecar
replaces), or exclude the duplicate from `_get_payload_size`. Also write raw
bytes instead of Base64 in MessagePack mode.
##########
superset/sql/execution/executor.py:
##########
@@ -404,6 +408,118 @@ def execute_sql_with_cursor(
return results
+def _decode_async_results_payload(blob: bytes) -> dict[str, Any]:
+ """
+ Decode a results backend blob written by the async SQL execution task.
+
+ The task stores a zlib-compressed payload serialized either as MessagePack
+ or as UTF-8 JSON, depending on ``RESULTS_BACKEND_USE_MSGPACK``. The format
+ is detected from the payload itself (a JSON object starts with ``{``, which
+ is never the first byte of a MessagePack map), so results stay readable if
+ the setting differs between the writing worker and the reading process.
+
+ :param blob: Compressed payload read from the results backend
+ :returns: The deserialized payload
+ """
+ import msgpack
+
+ from superset.utils import json
+
+ decompressed = cast(bytes, utils.zlib_decompress(blob, decode=False))
+ if decompressed.lstrip()[:1] == b"{":
+ return json.loads(decompressed)
+ return msgpack.loads(decompressed, raw=False)
+
+
+def _decode_async_statement_data(stmt_data: dict[str, Any]) -> pd.DataFrame |
None:
+ """
+ Convert one stored statement's data into a DataFrame.
+
+ SELECT results are Arrow IPC bytes (MessagePack) or Base64 strings (JSON).
+ Older JSON payloads contain records; DML statements store ``None``. A
+ SELECT returning no rows yields an empty DataFrame that keeps its columns,
+ matching the synchronous execution path.
+
+ :param stmt_data: Statement entry from the stored payload
+ :returns: DataFrame of the statement rows, or ``None`` for DML
+ """
+ data = stmt_data.get("data")
+ if data is None:
+ return None
+
+ if isinstance(data, str):
+ data = base64.b64decode(data)
+ if isinstance(data, (bytes, bytearray, memoryview)):
+ from superset.result_set import SupersetResultSet
+
+ table = pa.ipc.open_stream(pa.BufferReader(data)).read_all()
+ df = SupersetResultSet.convert_table_to_df(table)
+ # Older sidecars contain untyped values; do not guess lost types.
+ nested_columns = {
+ **stmt_data.get("nested_columns", {}),
+
**deserialize_nested_columns(stmt_data.get("nested_columns_msgpack", {})),
+ }
+ for name, values in nested_columns.items():
+ df[name] = values
Review Comment:
**🟡 Medium — An unknown sidecar tag fails the whole result**
`deserialize_nested_columns` raises on any tag it doesn't know
(`serialization.py:69`), and nothing here catches it. `get_result()` then
returns FAILED with no statements, even though the Arrow text for that column
is right here in `df`. Nothing writes unknown tags yet. But the first new tag
(e.g. one for `date`) would break every reader that's one version behind during
a rolling deploy.
Fix: decode each sidecar column in its own try/except and keep the Arrow
string column on failure. `test_nested_sidecar_rejects_unknown_type_tag` can
stay as a test of the low-level function.
##########
superset/sql/execution/executor.py:
##########
@@ -1084,26 +1200,38 @@ def _get_from_cache(self, sql: str, opts: QueryOptions)
-> QueryResult | None:
if cache_key is None:
return None
- if (cached := cache_manager.data_cache.get(cache_key)) is not None:
- # Reconstruct statement results from cached data
- statements = [
- StatementResult(
- original_sql=stmt_data["original_sql"],
- executed_sql=stmt_data["executed_sql"],
- data=stmt_data["data"],
- row_count=stmt_data["row_count"],
- execution_time_ms=stmt_data["execution_time_ms"],
- truncated=stmt_data.get("truncated", False),
+ try:
+ if (cached := cache_manager.data_cache.get(cache_key)) is not None:
+ # Reconstruct statement results from cached data
+ statements = [
+ StatementResult(
+ original_sql=stmt_data["original_sql"],
+ executed_sql=stmt_data["executed_sql"],
+ data=(
+ _records_to_dataframe(
+ stmt_data["data"],
+ stmt_data.get("columns", []),
+ dtypes=stmt_data.get("dtypes"),
+ )
+ if isinstance(stmt_data["data"], list)
+ else stmt_data["data"]
+ ),
+ row_count=stmt_data["row_count"],
+ execution_time_ms=stmt_data["execution_time_ms"],
+ truncated=stmt_data.get("truncated", False),
+ )
+ for stmt_data in cached.get("statements", [])
+ ]
+ return QueryResultType(
+ status=QueryStatus.SUCCESS,
+ statements=statements,
+ query_id=None,
+ is_cached=True,
+
total_execution_time_ms=cached.get("total_execution_time_ms", 0),
)
- for stmt_data in cached.get("statements", [])
- ]
- return QueryResultType(
- status=QueryStatus.SUCCESS,
- statements=statements,
- query_id=None,
- is_cached=True,
- total_execution_time_ms=cached.get("total_execution_time_ms",
0),
- )
+ except Exception: # pylint: disable=broad-except
+ # A stale or unreadable cache entry must not prevent query
execution.
+ logger.warning("Unable to decode cached SQL result; treating as a
miss")
Review Comment:
**🟢 Low — Cache-miss handler hides real errors**
This `except Exception` also wraps `cache_manager.data_cache.get()`. It logs
a fixed warning with no exception, so a Redis `ConnectionError` or a bug in
`_records_to_dataframe` looks just like a stale entry, and caching silently
turns into re-execution. On master, a `get()` failure returned FAILED.
Fix: log the exception. Consider labelling cache-backend errors separately
from decode errors.
```suggestion
logger.warning(
"Unable to decode cached SQL result; treating as a miss",
exc_info=True,
)
```
##########
superset/sql/execution/serialization.py:
##########
@@ -0,0 +1,105 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Lossless, transport-independent sidecars for supported nested SQL values."""
+
+import base64
+from datetime import datetime, timezone
+from decimal import Decimal
+from typing import Any
+from uuid import UUID
+
+import msgpack
+
+
+def _encode_nested_scalar(value: Any) -> msgpack.ExtType:
+ """Encode supported non-MessagePack scalars with explicit type tags."""
+ if type(value) is int:
+ return msgpack.ExtType(1, str(value).encode("ascii"))
+ if type(value) is datetime and (
+ value.tzinfo is None or type(value.tzinfo) is timezone
+ ):
+ return msgpack.ExtType(2, value.isoformat().encode("ascii"))
+ if type(value) is UUID:
+ return msgpack.ExtType(3, value.bytes)
+ if type(value) is Decimal:
+ return msgpack.ExtType(4, str(value).encode("ascii"))
+ raise TypeError("Unsupported nested SQL value")
+
+
+def _encode_nested_value(value: Any) -> Any:
+ """Walk exact supported types, excluding user-supplied extension tags."""
+ if type(value) is list:
+ return [_encode_nested_value(item) for item in value]
+ if type(value) is dict:
+ return {
+ _encode_nested_value(key): _encode_nested_value(item)
+ for key, item in value.items()
+ }
+ if type(value) is int and -(2**63) <= value < 2**64:
+ return value
+ if type(value) in (type(None), bool, float, str, bytes):
+ return value
+ return _encode_nested_scalar(value)
Review Comment:
**🟡 Medium — Sidecar silently skips common nested types, for the whole
column**
This accepts exact types only. `serialize_nested_columns` (`:89-90`) drops
the whole column, without logging, at the first unsupported value. These fall
back to JSON text:
- nested `date`/`time`/`timedelta` (Postgres `date[]`/`interval[]`, BigQuery
`ARRAY<DATE>`, a STRUCT with a DATE)
- pytz/`ZoneInfo` datetimes (e.g. Trino named zones)
- numpy/`pd.Timestamp` values
For example, sync returns `[[date(2026, 1, 1)]]` and async returns
`['["2026-01-01"]']`. Because the fallback is per column, one such value also
turns neighbouring rows like `[1, 2]` into `'[1, 2]'`. `mcp.md:42-45` doesn't
list date/time/timedelta or numpy/pandas scalars among the fallbacks.
Fix: once the reader tolerates unknown tags, add tags for
`date`/`time`/`timedelta`. Log at debug when a column falls back, and list the
remaining fallbacks in the docs.
##########
superset/sql/execution/executor.py:
##########
@@ -404,6 +408,118 @@ def execute_sql_with_cursor(
return results
+def _decode_async_results_payload(blob: bytes) -> dict[str, Any]:
+ """
+ Decode a results backend blob written by the async SQL execution task.
+
+ The task stores a zlib-compressed payload serialized either as MessagePack
+ or as UTF-8 JSON, depending on ``RESULTS_BACKEND_USE_MSGPACK``. The format
+ is detected from the payload itself (a JSON object starts with ``{``, which
+ is never the first byte of a MessagePack map), so results stay readable if
+ the setting differs between the writing worker and the reading process.
+
+ :param blob: Compressed payload read from the results backend
+ :returns: The deserialized payload
+ """
+ import msgpack
+
+ from superset.utils import json
+
+ decompressed = cast(bytes, utils.zlib_decompress(blob, decode=False))
+ if decompressed.lstrip()[:1] == b"{":
+ return json.loads(decompressed)
+ return msgpack.loads(decompressed, raw=False)
+
+
+def _decode_async_statement_data(stmt_data: dict[str, Any]) -> pd.DataFrame |
None:
+ """
+ Convert one stored statement's data into a DataFrame.
+
+ SELECT results are Arrow IPC bytes (MessagePack) or Base64 strings (JSON).
+ Older JSON payloads contain records; DML statements store ``None``. A
+ SELECT returning no rows yields an empty DataFrame that keeps its columns,
+ matching the synchronous execution path.
+
+ :param stmt_data: Statement entry from the stored payload
+ :returns: DataFrame of the statement rows, or ``None`` for DML
+ """
+ data = stmt_data.get("data")
+ if data is None:
+ return None
+
+ if isinstance(data, str):
+ data = base64.b64decode(data)
+ if isinstance(data, (bytes, bytearray, memoryview)):
+ from superset.result_set import SupersetResultSet
+
+ table = pa.ipc.open_stream(pa.BufferReader(data)).read_all()
+ df = SupersetResultSet.convert_table_to_df(table)
+ # Older sidecars contain untyped values; do not guess lost types.
+ nested_columns = {
+ **stmt_data.get("nested_columns", {}),
+
**deserialize_nested_columns(stmt_data.get("nested_columns_msgpack", {})),
+ }
Review Comment:
**🟢 Low — `nested_columns` fallback is dead compatibility code again**
The plain `nested_columns` key was only written by intermediate commit
98a05a7, as `temporal_schema` was last round. Code that still serves it:
- the merge here
- `test_legacy_nested_sidecar_remains_readable`
(`test_async_results.py:1031`)
- "untyped nested sidecars remain readable" (`mcp.md:53`)
Fix: drop the fallback (suggestion below), that test, and the docs phrase.
```suggestion
nested_columns = deserialize_nested_columns(
stmt_data.get("nested_columns_msgpack", {})
)
```
##########
superset/sql/execution/serialization.py:
##########
@@ -0,0 +1,105 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Lossless, transport-independent sidecars for supported nested SQL values."""
+
+import base64
+from datetime import datetime, timezone
+from decimal import Decimal
+from typing import Any
+from uuid import UUID
+
+import msgpack
+
+
+def _encode_nested_scalar(value: Any) -> msgpack.ExtType:
+ """Encode supported non-MessagePack scalars with explicit type tags."""
+ if type(value) is int:
+ return msgpack.ExtType(1, str(value).encode("ascii"))
+ if type(value) is datetime and (
+ value.tzinfo is None or type(value.tzinfo) is timezone
+ ):
+ return msgpack.ExtType(2, value.isoformat().encode("ascii"))
+ if type(value) is UUID:
+ return msgpack.ExtType(3, value.bytes)
+ if type(value) is Decimal:
+ return msgpack.ExtType(4, str(value).encode("ascii"))
+ raise TypeError("Unsupported nested SQL value")
+
+
+def _encode_nested_value(value: Any) -> Any:
+ """Walk exact supported types, excluding user-supplied extension tags."""
+ if type(value) is list:
+ return [_encode_nested_value(item) for item in value]
+ if type(value) is dict:
+ return {
+ _encode_nested_value(key): _encode_nested_value(item)
+ for key, item in value.items()
+ }
+ if type(value) is int and -(2**63) <= value < 2**64:
+ return value
+ if type(value) in (type(None), bool, float, str, bytes):
+ return value
+ return _encode_nested_scalar(value)
+
+
+def _decode_nested_scalar(code: int, data: bytes) -> Any:
+ """Restore only the scalar types explicitly supported by the sidecar."""
+ if code == 1:
+ return int(data)
+ if code == 2:
+ return datetime.fromisoformat(data.decode("ascii"))
+ if code == 3:
+ return UUID(bytes=data)
+ if code == 4:
+ return Decimal(data.decode("ascii"))
+ raise ValueError("Unsupported nested SQL type tag")
+
+
+def serialize_nested_columns(columns: dict[str, list[Any]]) -> dict[str, str]:
+ """Encode nested columns as Base64 MessagePack, or keep Arrow's fallback.
+
+ Explicit extensions preserve arbitrary-sized integers, datetime, UUID and
+ Decimal; MessagePack retains bytes without text decoding. Strict types
+ prevent silent conversions (for example, tuples into lists). A column
+ containing unsupported values is omitted so its Arrow string column is
+ used instead. Sidecars never rely on executable object deserialization.
+ """
+ encoded: dict[str, str] = {}
+ for name, values in columns.items():
+ try:
+ data = msgpack.dumps(
+ _encode_nested_value(values),
+ use_bin_type=True,
+ strict_types=True,
+ )
+ except (TypeError, ValueError, OverflowError):
Review Comment:
**🟢 Low — Deep nesting fails the async query**
`_encode_nested_value` is recursive, and `RecursionError` isn't in this
except list. A value nested about 500 levels deep raises in the writer and
fails the whole async query. Sync returns the same value, and master's writer
stored it. It's a pathological case, but the fix is one line.
```suggestion
except (TypeError, ValueError, OverflowError, RecursionError):
```
##########
docs/developer_docs/extensions/mcp.md:
##########
@@ -28,6 +28,34 @@ under the License.
Model Context Protocol (MCP) integration allows extensions to register custom
AI agent capabilities that integrate seamlessly with Superset's MCP service.
Extensions can provide both **tools** (executable functions) and **prompts**
(interactive guidance) that AI agents can discover and use.
+## Python SQL query results
Review Comment:
**🟢 Low — Docs section sits on the MCP page**
This section documents `Database.execute_async()` above "What is MCP?", but
nothing in it is MCP-specific. The "Backend APIs" section in
`docs/developer_docs/extensions/development.md` is a closer fit.
##########
superset/sql/execution/executor.py:
##########
@@ -404,6 +408,118 @@ def execute_sql_with_cursor(
return results
+def _decode_async_results_payload(blob: bytes) -> dict[str, Any]:
+ """
+ Decode a results backend blob written by the async SQL execution task.
+
+ The task stores a zlib-compressed payload serialized either as MessagePack
+ or as UTF-8 JSON, depending on ``RESULTS_BACKEND_USE_MSGPACK``. The format
+ is detected from the payload itself (a JSON object starts with ``{``, which
+ is never the first byte of a MessagePack map), so results stay readable if
+ the setting differs between the writing worker and the reading process.
+
+ :param blob: Compressed payload read from the results backend
+ :returns: The deserialized payload
+ """
+ import msgpack
+
+ from superset.utils import json
Review Comment:
**🟢 Low — Local imports remain**
`import msgpack` and `from superset.utils import json` don't need to be
local. This module already imports `serialization` (which imports msgpack) at
module level, and there's no import cycle to avoid.
Fix: move both to module level.
--
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]
