[
https://issues.apache.org/jira/browse/FLINK-41295?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136389#comment-18136389
]
Sangkyoon Nam commented on FLINK-41295:
---------------------------------------
Thanks for the precise report. Unless you plan to fix this yourself, could a
committer assign it to me?
I reproduced it on master with the Cython coder. I added three regression tests
to {{test_coders.py}} with values from 0001-01-02 to 9999-12-30, including both
int32 boundaries. Their Cython variants failed in 3 of 3 runs with the current
{{coder_impl_fast.pyx}} (0001-01-02 decodes as 1906-06-08 18:35:44), and all of
them passed in 3 of 3 runs after declaring {{seconds}} as {{int64_t}}. The pure
Python coder already decodes these values correctly. The fix also covers
{{LocalZonedTimestampCoderImpl}}, since it decodes through the same method. I
haven't run the Table API UDF and UDAF scripts from the description, because I
don't have a Flink distribution set up here.
> PyFlink: Python UDFs receive TIMESTAMP values before 1901-12-13 or after
> 2038-01-19 as wrong dates (int32 seconds in the Cython TimestampCoderImpl)
> ---------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41295
> URL: https://issues.apache.org/jira/browse/FLINK-41295
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 2.0.1, 2.1.1, 2.3.0
> Reporter: Yaoxuan Wu
> Priority: Major
> Labels: correctness
>
> {code:python}
> import datetime as dt
> from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as
> D
> from pyflink.table.expressions import col
> from pyflink.table.udf import udf
> t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
> t = t_env.from_elements([(dt.datetime(1, 1, 2),), (dt.datetime(2000, 1, 1),),
> (dt.datetime(2300, 1, 1),)],
> D.ROW([D.FIELD('ts', D.TIMESTAMP(3))]))
> show = udf(lambda x: str(x), result_type=D.STRING())
> t.select(col('ts'), show(col('ts')).alias('in_python')).execute().print()
> # expected: in_python = 0001-01-02 00:00:00, 2000-01-01 00:00:00, 2300-01-01
> 00:00:00
> # actual: in_python = 1906-06-08 18:35:44, 2000-01-01 00:00:00, 2027-10-18
> 11:03:28
> {code}
> The same happens to the grouping keys returned by a general Python UDAF:
> {{GROUP BY ts}} with a Python {{AggregateFunction}} in streaming mode returns
> the keys 0001-01-02 and 9999-12-30 23:59:59 as 1906-06-08 18:35:44 and
> 1969-12-22 02:12:15, while the native {{COUNT}} returns the original keys.
> Cause: {{TimestampCoderImpl._decode_timestamp_data_from_stream}}
> (pyflink/fn_execution/coder_impl_fast.pyx) stores the epoch seconds in a
> 32-bit variable:
> {code}
> cdef int64_t milliseconds
> cdef int32_t nanoseconds, seconds, microseconds
> ...
> seconds = milliseconds // 1000
> return
> datetime.datetime.utcfromtimestamp(seconds).replace(microsecond=microseconds)
> {code}
> Epoch seconds outside [-2^31, 2^31) wrap (0001-01-02 is -62135510400 s, which
> wraps to 1906-06-08 18:35:44). {{LocalZonedTimestampCoderImpl}} extends this
> class and decodes through the same method.
> Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
