[
https://issues.apache.org/jira/browse/FLINK-41196?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Biao Geng updated FLINK-41196:
------------------------------
Description:
h3. Problem
Pandas conversions and regular Python UDFs in process mode incorrectly
interpret UTC timestamps as local time.
With {{Asia/Shanghai}}, {{00:00 UTC}} becomes {{00:00+08:00}} instead of
{{08:00+08:00}}. Encoding timezone-aware values also fails to preserve the
instant. The two errors can cancel out in round-trip tests.
Affected paths include {{Table.to_pandas()}},
{{TableEnvironment.from_pandas()}}, pandas UDFs, and regular Python UDFs in
process mode. 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. Proposed fix
Decode values as UTC before converting to the session time zone; normalize
timezone-aware values to UTC before encoding. Preserve existing handling of
timezone-naive inputs.
Cover shared pandas helpers and both slow/fast Python UDF coders with
independent encoding/decoding and batch/streaming tests, including DST
transitions. Thread mode is out of scope.
h3. Environment
Originally reproduced 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.
was:
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.
> 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
> Pandas conversions and regular Python UDFs in process mode incorrectly
> interpret UTC timestamps as local time.
> With {{Asia/Shanghai}}, {{00:00 UTC}} becomes {{00:00+08:00}} instead of
> {{08:00+08:00}}. Encoding timezone-aware values also fails to preserve the
> instant. The two errors can cancel out in round-trip tests.
> Affected paths include {{Table.to_pandas()}},
> {{TableEnvironment.from_pandas()}}, pandas UDFs, and regular Python UDFs in
> process mode. 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. Proposed fix
> Decode values as UTC before converting to the session time zone; normalize
> timezone-aware values to UTC before encoding. Preserve existing handling of
> timezone-naive inputs.
> Cover shared pandas helpers and both slow/fast Python UDF coders with
> independent encoding/decoding and batch/streaming tests, including DST
> transitions. Thread mode is out of scope.
> h3. Environment
> Originally reproduced 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)
