aminghadersohi commented on code in PR #45348:
URL: https://github.com/apache/superset/pull/45348#discussion_r4290933699
##########
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:
Fixed cache diagnostics with separate backend/decode warnings carrying
exception information in 3c7c8adf04a9f2a26012a4913289da1437f33592. Regression:
tests/unit_tests/sql/execution/test_async_results.py::test_cache_failure_logs_exception_and_source
(the named cases fail with source at d205c6fd3ac016d58b688b3eec250f974491eb0c
in a detached scratch worktree and pass at this commit).
Siblings checked:
- superset/sql/execution/executor.py:550 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/executor.py:686 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/executor.py:1055 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/executor.py:1196 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/executor.py:1260 — not affected: cache writes do
not use the cache-read exception handler
##########
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:
Fixed RecursionError fallback in the optional nested sidecar writer in
3c7c8adf04a9f2a26012a4913289da1437f33592. Regression:
tests/unit_tests/sql/execution/test_async_results.py::test_deep_nested_sidecar_falls_back_to_arrow
(the named cases fail with source at d205c6fd3ac016d58b688b3eec250f974491eb0c
in a detached scratch worktree and pass at this commit).
Siblings checked:
- superset/sql/execution/serialization.py:90 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/celery_task.py:171 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/executor.py:431 — not affected: an omitted sidecar
leaves the Arrow text intact in either transport
- superset/result_set.py:369 — not affected: sync uses the original objects
without the recursive sidecar encoder
##########
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:
Fixed the stale compatibility path by removing the unreleased untyped
nested_columns fallback, its compatibility test and the docs claim in
3c7c8adf04a9f2a26012a4913289da1437f33592. Regression:
tests/unit_tests/sql/execution/test_async_results.py::test_intermediate_nested_sidecar_is_ignored
(the named cases fail with source at d205c6fd3ac016d58b688b3eec250f974491eb0c
in a detached scratch worktree and pass at this commit).
Siblings checked:
- superset/sql/execution/executor.py:431 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- docs/developer_docs/extensions/development.md:234 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/celery_task.py:171 — not affected: the writer only
emits nested_columns_msgpack
- superset/sql/execution/executor.py:483 — not affected: released
JSON-record payloads still use the records decoder, not the removed sidecar key
--
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]
