[
https://issues.apache.org/jira/browse/FLINK-41196?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Biao Geng updated FLINK-41196:
------------------------------
Summary: PyFlink pandas conversions and regular Python UDFs change
TIMESTAMP_LTZ instants in non-UTC session time zones (was: PyFlink
Table.to_pandas() changes TIMESTAMP_LTZ instants in non-UTC session time zones)
> PyFlink pandas conversions and regular Python UDFs change TIMESTAMP_LTZ
> instants in non-UTC session time zones
> --------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41196
> URL: https://issues.apache.org/jira/browse/FLINK-41196
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Reporter: Biao Geng
> Priority: Major
> Labels: pull-request-available
>
> h3. Problem
> {{Table.to_pandas()}} incorrectly interprets UTC timestamp values as local
> wall-clock time when converting {{TIMESTAMP_LTZ}} columns to pandas.
> For example, with {{Asia/Shanghai}}, an instant at {{00:00 UTC}} becomes
> {{00:00+08:00}} instead of {{08:00+08:00}}, shifting the actual instant eight
> hours earlier.
> This issue predates FLINK-41032.
> h3. Reproducer
> {code:python}
> import pandas as pd
> from pyflink.table import EnvironmentSettings, TableEnvironment
> t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
> t_env.get_config().set_local_timezone("Asia/Shanghai")
> actual = t_env.sql_query("""
> SELECT TO_TIMESTAMP_LTZ(1704067200123, 3) AS ts
> """).to_pandas()["ts"].iloc[0]
> assert actual == pd.Timestamp("2024-01-01 00:00:00.123", tz="UTC")
> {code}
> *Expected:* {{2024-01-01 08:00:00.123+08:00}}
> *Actual:* {{2024-01-01 00:00:00.123+08:00}}
> h3. Cause and proposed fix
> The {{to_pandas()}} path calls {{tz_convert_from_internal()}}, which applies
> {{tz_localize(local_tz)}} to timestamp values representing UTC.
> This path should instead use:
> {code:python}
> s.dt.tz_localize("UTC").dt.tz_convert(local_tz)
> {code}
> Check other callers before changing the shared helper globally. Add
> regression tests verifying instant preservation in batch and streaming modes.
> h3. Environment
> Verified with pandas 2.2.3, PyArrow 16.1.0 and a local Flink 2.3-SNAPSHOT
> Java build, including the Python implementation before FLINK-41032.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
