ChahatKumar commented on code in PR #73783:
URL: https://github.com/apache/airflow/pull/73783#discussion_r4323596412
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -176,14 +180,44 @@ def _new_encoder() -> msgspec.msgpack.Encoder:
return msgspec.msgpack.Encoder(enc_hook=_msgpack_enc_hook)
+def _encode_large_ints(value: Any) -> Any:
+ if isinstance(value, int) and not -(2**63) <= value < 2**64:
+ return msgspec.msgpack.Ext(
+ MP_HUGEINT, value.to_bytes((value.bit_length() + 8) // 8, "big",
signed=True)
+ )
+ if isinstance(value, dict):
+ return {key: _encode_large_ints(item) for key, item in value.items()}
+ if isinstance(value, (list, tuple)):
+ return [_encode_large_ints(item) for item in value]
+ if isinstance(value, msgspec.Struct):
+ return msgspec.structs.replace(
+ value, **{name: _encode_large_ints(getattr(value, name)) for name
in value.__struct_fields__}
+ )
+ if isinstance(value, BaseModel):
+ return _encode_large_ints(value.model_dump(exclude_unset=True))
Review Comment:
The oversized values were already nested in containers for example, inside
DagRun.conf within the task-startup response so container handling is required
to fix the actual failure with current approach, it seems.
enc_hook is msgspec’s callback for encoding unsupported custom Python
objects, but it is not called for an oversized built-in int; msgspec raises
OverflowError directly, hence the need to catch that error, recursively replace
nested oversized integers with MP_HUGEINT extensions, and retry encoding.
@dhkim1920 can you add your thoughts on this as well and address the latest
comments
--
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]
