Yaoxuan Wu created FLINK-41295:
----------------------------------
Summary: 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.3.0, 2.1.1, 2.0.1
Reporter: Yaoxuan Wu
{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)
