[
https://issues.apache.org/jira/browse/FLINK-35273?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18111145#comment-18111145
]
Bhanu Chander Vallabaneni commented on FLINK-35273:
---------------------------------------------------
Still reproducible on {{master}} as far as I can tell from the code, and I have
a specific
hypothesis — posting it here in case it saves whoever picks this up some time.
I would be glad to
take the ticket if a committer can assign it, but the diagnosis is the useful
part either way.
The Python side does honour a timezone when it decodes a TIMESTAMP_LTZ:
{code:python}
class LocalZonedTimestampCoderImpl(TimestampCoderImpl):
def __init__(self, precision, timezone):
...
self.timezone = timezone
def internal_to_timestamp(self, milliseconds, nanoseconds):
return self.timezone.localize(
super().internal_to_timestamp(milliseconds, nanoseconds))
{code}
but that {{timezone}} does not come from the table config. In
{{pyflink/fn_execution/coders.py}} it is built from an environment variable:
{code:python}
timezone = pytz.timezone(os.environ['TABLE_LOCAL_TIME_ZONE'])
{code}
So the value that reaches the coder depends on {{TABLE_LOCAL_TIME_ZONE}} being
set in the Python
worker's environment from {{table.local-time-zone}}. My reading is that the
reported symptom is that
variable not being propagated on this path, so the coder localises against
whatever the fallback is
rather than the {{set_local_timezone("UTC")}} the user asked for — which
matches the output being
consistently off by the difference between the configured zone and the
machine's.
If that is right, the fix belongs where the worker environment is prepared
rather than in the coder,
and the test wants to assert the decoded value for two different configured
zones with the machine
zone held constant.
To be clear about what I have and have not done: this is from reading
{{master}}, not from running
it — I do not have a Flink build here yet. Worth confirming before acting on it.
> PyFlink's LocalZonedTimestampType should respect timezone set by
> set_local_timezone
> -----------------------------------------------------------------------------------
>
> Key: FLINK-35273
> URL: https://issues.apache.org/jira/browse/FLINK-35273
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Reporter: Biao Geng
> Priority: Major
>
> The issue is from
> https://apache-flink.slack.com/archives/C065944F9M2/p1714134880878399
> When using TIMESTAMP_LTZ in PyFlink while setting a different time zone, it
> turns out that the output result does not show the expected result.
> Here is my test codes:
> {code:python}
> from pyflink.datastream import StreamExecutionEnvironment
> from pyflink.common import Types, Configuration
> from pyflink.table import DataTypes, StreamTableEnvironment
> from datetime import datetime
> import pytz
> config = Configuration()
> config.set_string("python.client.executable",
> "/usr/local/Caskroom/miniconda/base/envs/myenv/bin/python")
> config.set_string("python.executable",
> "/usr/local/Caskroom/miniconda/base/envs/myenv/bin/python")
> env = StreamExecutionEnvironment.get_execution_environment(config)
> t_env = StreamTableEnvironment.create(env)
> t_env.get_config().set_local_timezone("UTC")
> # t_env.get_config().set_local_timezone("GMT-08:00")
> input_table = t_env.from_elements(
> [
> (
> "elementA",
> datetime(year=2024, month=4, day=12, hour=8, minute=35),
> ),
> (
> "elementB",
> datetime(year=2024, month=4, day=12, hour=8, minute=35,
> tzinfo=pytz.utc),
> # datetime(year=2024, month=4, day=12, hour=8, minute=35,
> tzinfo=pytz.timezone('America/New_York')),
> ),
> ],
> DataTypes.ROW(
> [
> DataTypes.FIELD("name", DataTypes.STRING()),
> DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)),
> ]
> ),
> )
> input_table.execute().print()
> # SQL
> sql_result = t_env.execute_sql("CREATE VIEW MyView1 AS SELECT
> TO_TIMESTAMP_LTZ(1712910900000, 3);")
> t_env.execute_sql("CREATE TABLE Sink (`t` TIMESTAMP_LTZ) WITH
> ('connector'='print');")
> t_env.execute_sql("INSERT INTO Sink SELECT * FROM MyView1;")
> {code}
> The output is:
> {code:java}
> +----+--------------------------------+-------------------------+
> | op | name | timestamp |
> +----+--------------------------------+-------------------------+
> | +I | elementA | 2024-04-12 08:35:00.000 |
> | +I | elementB | 2024-04-12 16:35:00.000 |
> +----+--------------------------------+-------------------------+
> 2 rows in set
> +I[2024-04-12T08:35:00Z]
> {code}
> In pyflink/tables/types.py, the `LocalZonedTimestampType` class will use
> follow logic to convert python obj to sql type:
> {code:python}
> EPOCH_ORDINAL = calendar.timegm(time.localtime(0)) * 10 ** 6
> ...
> def to_sql_type(self, dt):
> if dt is not None:
> seconds = (calendar.timegm(dt.utctimetuple()) if dt.tzinfo
> else time.mktime(dt.timetuple()))
> return int(seconds) * 10 ** 6 + dt.microsecond +
> self.EPOCH_ORDINAL
> {code}
> It shows that the EPOCH_ORDINAL is calculated when the PVM starts but is not
> decided by the timezone set by `set_local_timezone`.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
