aminghadersohi commented on code in PR #45348:
URL: https://github.com/apache/superset/pull/45348#discussion_r4290932679
##########
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:
Fixed the size-limit double charge and MessagePack Base64 wrapper in
3c7c8adf04a9f2a26012a4913289da1437f33592. Regression:
tests/unit_tests/sql/execution/test_async_results.py::test_nested_payload_limit_does_not_charge_duplicate_sidecar
and test_msgpack_nested_sidecar_uses_raw_bytes (the named cases fail with
source at d205c6fd3ac016d58b688b3eec250f974491eb0c in a detached scratch
worktree and pass at this commit).
Remaining: storage retains Arrow text plus a typed sidecar for per-column
fallback; the docs explicitly disclose both copies. The limit uses the
suggested exclusion option.
Siblings checked:
- superset/sql/execution/celery_task.py:257 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/serialization.py:90 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/executor.py:431 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql_lab.py:714 — not affected: separate SQL Lab payload has no
nested sidecars
- superset/sql_lab.py:803 — not affected: return_results serializes the
separate SQL Lab payload, without sidecars
##########
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:
Fixed per-column fallback for unknown tags, malformed sidecars and
mismatched row counts in 3c7c8adf04a9f2a26012a4913289da1437f33592. Regression:
tests/unit_tests/sql/execution/test_async_results.py::test_bad_nested_sidecar_keeps_arrow_and_other_columns
(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
- superset/sql/execution/executor.py:1501 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/serialization.py:118 — not affected: the low-level
decoder intentionally remains strict;
test_nested_sidecar_rejects_unknown_type_tag pins this
- superset/result_set.py:369 — not affected: sync restores original Python
values without decoding sidecars
##########
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:
Fixed nested date/time/timedelta round-trips and added debug logging for
unsupported-column fallback in 3c7c8adf04a9f2a26012a4913289da1437f33592.
Regression:
tests/unit_tests/sql/execution/test_async_results.py::test_nested_temporal_scalars_match_sync
and test_unsupported_nested_sidecar_logs_fallback (the named cases fail with
source at d205c6fd3ac016d58b688b3eec250f974491eb0c in a detached scratch
worktree and pass at this commit).
Remaining: named-timezone datetimes/times, numpy/pandas scalar subclasses,
sets and tuples retain whole-column Arrow-string fallback; development.md lists
these limitations.
Siblings checked:
- superset/sql/execution/serialization.py:31 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/celery_task.py:171 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/sql/execution/executor.py:431 — fixed in
3c7c8adf04a9f2a26012a4913289da1437f33592
- superset/result_set.py:369 — not affected: sync retains the original
objects and does not use sidecar tags
- superset/sql/execution/executor.py:1196 — not affected: record caches
preserve object dtype and do not serialize nested sidecars
--
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]
