Dian Fu created FLINK-41484:
-------------------------------
Summary: Fix TIMESTAMP_LTZ handling in nested Pandas conversions
and thread-mode Python UDFs
Key: FLINK-41484
URL: https://issues.apache.org/jira/browse/FLINK-41484
Project: Flink
Issue Type: Bug
Components: API / Python
Reporter: Dian Fu
TIMESTAMP_LTZ conversion remains incomplete or incorrect in two areas:
1. Nested values in PyFlink pandas/Arrow conversion paths, including
Table.to_pandas() and pandas UDFs.
2. Both top-level and nested values in regular Python UDFs running with
python.execution-mode=thread.
These paths should preserve the represented instant and expose supported
TIMESTAMP_LTZ inputs as timezone-aware Python values in the session time zone.
1. Nested TIMESTAMP_LTZ in pandas/Arrow paths
The shared pandas conversion logic handles top-level TIMESTAMP_LTZ columns but
does not consistently apply the same conversion recursively to nested values.
Depending on the API and schema, nested TIMESTAMP_LTZ is either explicitly
rejected or accepted without the required time-zone conversion.
For example, with table.local-time-zone=Asia/Shanghai and both values
representing epoch 0, decoding an Arrow batch for a pandas UDF produces:
Top-level TIMESTAMP_LTZ:
Timestamp('1970-01-01 08:00:00+0800', tz='Asia/Shanghai')
Value inside MAP<STRING, TIMESTAMP_LTZ>:
[('k', datetime.datetime(1970, 1, 1, 0, 0))]
The nested value has no tzinfo and retains UTC calendar fields. A UDF reading
the hour therefore observes 8 for the top-level value and 0 for the nested
value, although both represent the same instant and have the same logical
timestamp type.
In the pandas UDF path:
- ARRAY<TIMESTAMP_LTZ> and ROW fields directly containing TIMESTAMP_LTZ are
explicitly rejected.
- MAP keys/values containing TIMESTAMP_LTZ can bypass these checks and reach
Python without time-zone conversion.
- The same gap can occur when such a MAP is nested inside another ARRAY or ROW.
The updated PR #29297 adds rejection checks for certain nested LTZ schemas in
Table.to_pandas(). This prevents those schemas from being silently misconverted
on that path, but does not provide recursive LTZ support or fix the accepted
MAP schemas in pandas UDFs.
The confirmed silent-conversion defect is on the pandas input/decode side.
Encoding must also be covered by regression tests; some timezone-aware nested
outputs are already normalized correctly by PyArrow.
2. TIMESTAMP_LTZ in regular Python UDF thread mode
With python.execution-mode=thread, top-level and nested LTZ values follow
different conversion paths, both of which have problems.
For top-level TIMESTAMP_LTZ, with session time zone Asia/Shanghai:
- Input epoch 0 becomes datetime.datetime(1970, 1, 1, 0, 0), with tzinfo=None.
- The expected input is 1970-01-01 08:00:00+08:00.
- Returning datetime.datetime(1970, 1, 1, 8,
tzinfo=datetime.timezone(datetime.timedelta(hours=8))) produces epoch
milliseconds 28800000 instead of 0.
For nested TIMESTAMP_LTZ:
- ARRAY elements, ROW fields, and MAP keys/values reach Python as
pemja.PyJObject wrappers around java.time.Instant.
- Returning a newly constructed Python datetime inside ARRAY<TIMESTAMP_LTZ>
fails with ArrayStoreException: java.sql.Timestamp.
- Returning a Python datetime inside ROW or a MAP value fails with
ClassCastException: java.sql.Timestamp cannot be cast to java.time.Instant.
Identity round trips can still pass. A top-level identity UDF can preserve
epoch 0, and a nested identity UDF can return the original Java Instant
wrappers unchanged. Input representation and output instant preservation must
therefore be tested independently.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
