This is an automated email from the ASF dual-hosted git repository.
hxb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ef0489b5131 [FLINK-29648][python] Fix "LocalTimeTypeInfo not
supported" error
ef0489b5131 is described below
commit ef0489b5131d03112d8a4ef11962a2d1ba6a9f54
Author: Juntao Hu <[email protected]>
AuthorDate: Mon Oct 17 12:54:57 2022 +0800
[FLINK-29648][python] Fix "LocalTimeTypeInfo not supported" error
This closes #21079.
---
flink-python/pyflink/common/typeinfo.py | 46 ++++++++++++++++++++++
.../pyflink/fn_execution/coder_impl_fast.pxd | 10 +----
.../pyflink/fn_execution/coder_impl_fast.pyx | 16 ++++----
.../table/tests/test_table_environment_api.py | 17 ++++++++
4 files changed, 73 insertions(+), 16 deletions(-)
diff --git a/flink-python/pyflink/common/typeinfo.py
b/flink-python/pyflink/common/typeinfo.py
index c1ece76f9d9..861231ed167 100644
--- a/flink-python/pyflink/common/typeinfo.py
+++ b/flink-python/pyflink/common/typeinfo.py
@@ -720,6 +720,40 @@ class MapTypeInfo(TypeInformation):
return 'MapTypeInfo<{}, {}>'.format(self._key_type_info,
self._value_type_info)
+class LocalTimeTypeInfo(TypeInformation):
+
+ class TimeType(Enum):
+ LOCAL_DATE = 0
+ LOCAL_TIME = 1
+ LOCAL_DATE_TIME = 2
+
+ def __init__(self, time_type: TimeType):
+ super(LocalTimeTypeInfo, self).__init__()
+ self._time_type = time_type
+
+ def get_java_type_info(self) -> JavaObject:
+ if self._j_typeinfo is None:
+ jvm = get_gateway().jvm
+ if self._time_type == LocalTimeTypeInfo.TimeType.LOCAL_DATE:
+ self._j_typeinfo = \
+
jvm.org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo.LOCAL_DATE
+ elif self._time_type == LocalTimeTypeInfo.TimeType.LOCAL_TIME:
+ self._j_typeinfo = \
+
jvm.org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo.LOCAL_TIME
+ elif self._time_type == LocalTimeTypeInfo.TimeType.LOCAL_DATE_TIME:
+ self._j_typeinfo = \
+
jvm.org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo.LOCAL_DATE_TIME
+ else:
+ raise TypeError('Unsupported TimeType:
{}'.format(self._time_type.name))
+ return self._j_typeinfo
+
+ def __eq__(self, other):
+ return self.__class__ == other.__class__ and self._time_type ==
other._time_type
+
+ def __repr__(self):
+ return 'LocalTimeTypeInfo<{}>'.format(self._time_type.name)
+
+
class ExternalTypeInfo(TypeInformation):
def __init__(self, type_info: TypeInformation):
super(ExternalTypeInfo, self).__init__()
@@ -1062,6 +1096,18 @@ def _from_java_type(j_type_info: JavaObject) ->
TypeInformation:
j_element_type_info = j_type_info.getElementTypeInfo()
return ListTypeInfo(_from_java_type(j_element_type_info))
+ JLocalTimeTypeInfo =
gateway.jvm.org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo
+ if _is_instance_of(j_type_info, JLocalTimeTypeInfo):
+ if j_type_info.equals(JLocalTimeTypeInfo.LOCAL_DATE):
+ time_type = LocalTimeTypeInfo.TimeType.LOCAL_DATE
+ elif j_type_info.equals(JLocalTimeTypeInfo.LOCAL_TIME):
+ time_type = LocalTimeTypeInfo.TimeType.LOCAL_TIME
+ elif j_type_info.equals(JLocalTimeTypeInfo.LOCAL_DATE_TIME):
+ time_type = LocalTimeTypeInfo.TimeType.LOCAL_DATE_TIME
+ else:
+ raise TypeError("Unsupported LocalTimeTypeInfo: %s." %
j_type_info.toString())
+ return LocalTimeTypeInfo(time_type)
+
JExternalTypeInfo =
gateway.jvm.org.apache.flink.table.runtime.typeutils.ExternalTypeInfo
if _is_instance_of(j_type_info, JExternalTypeInfo):
TypeInfoDataTypeConverter = \
diff --git a/flink-python/pyflink/fn_execution/coder_impl_fast.pxd
b/flink-python/pyflink/fn_execution/coder_impl_fast.pxd
index e66a0da6d79..05d3ba5fd6b 100644
--- a/flink-python/pyflink/fn_execution/coder_impl_fast.pxd
+++ b/flink-python/pyflink/fn_execution/coder_impl_fast.pxd
@@ -194,16 +194,10 @@ cdef class AvroCoderImpl(FieldCoderImpl):
cdef object _writer
cdef class LocalDateCoderImpl(FieldCoderImpl):
- @staticmethod
- cdef _encode_to_stream(value, OutputStream out_stream)
- @staticmethod
- cdef _decode_from_stream(InputStream in_stream)
+ pass
cdef class LocalTimeCoderImpl(FieldCoderImpl):
- @staticmethod
- cdef _encode_to_stream(value, OutputStream out_stream)
- @staticmethod
- cdef _decode_from_stream(InputStream in_stream)
+ pass
cdef class LocalDateTimeCoderImpl(FieldCoderImpl):
pass
diff --git a/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
b/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
index 00ba1fd42c7..41ed4ab38c4 100644
--- a/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
+++ b/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
@@ -967,7 +967,7 @@ cdef class AvroCoderImpl(FieldCoderImpl):
cdef class LocalDateCoderImpl(FieldCoderImpl):
@staticmethod
- cdef _encode_to_stream(value, OutputStream out_stream):
+ def _encode_to_stream(value, OutputStream out_stream):
if value is None:
out_stream.write_int32(0xFFFFFFFF)
out_stream.write_int16(0xFFFF)
@@ -977,7 +977,7 @@ cdef class LocalDateCoderImpl(FieldCoderImpl):
out_stream.write_int8(value.day)
@staticmethod
- cdef _decode_from_stream(InputStream in_stream):
+ def _decode_from_stream(InputStream in_stream):
year = in_stream.read_int32()
if year == 0xFFFFFFFF:
in_stream.read(2)
@@ -987,15 +987,15 @@ cdef class LocalDateCoderImpl(FieldCoderImpl):
return datetime.date(year, month, day)
cpdef encode_to_stream(self, value, OutputStream out_stream):
- return LocalDateCoderImpl._encode_to_stream(value, out_stream)
+ return self._encode_to_stream(value, out_stream)
cpdef decode_from_stream(self, InputStream in_stream, size_t length):
- return LocalDateCoderImpl._decode_from_stream(in_stream)
+ return self._decode_from_stream(in_stream)
cdef class LocalTimeCoderImpl(FieldCoderImpl):
@staticmethod
- cdef _encode_to_stream(value, OutputStream out_stream):
+ def _encode_to_stream(value, OutputStream out_stream):
if value is None:
out_stream.write_int8(0xFF)
out_stream.write_int16(0xFFFF)
@@ -1007,7 +1007,7 @@ cdef class LocalTimeCoderImpl(FieldCoderImpl):
out_stream.write_int32(value.microsecond * 1000)
@staticmethod
- cdef _decode_from_stream(InputStream in_stream):
+ def _decode_from_stream(InputStream in_stream):
hour = in_stream.read_int8()
if hour == 0xFF:
in_stream.read(6)
@@ -1018,10 +1018,10 @@ cdef class LocalTimeCoderImpl(FieldCoderImpl):
return datetime.time(hour, minute, second, nano // 1000)
cpdef encode_to_stream(self, value, OutputStream out_stream):
- return LocalTimeCoderImpl._encode_to_stream(value, out_stream)
+ return self._encode_to_stream(value, out_stream)
cpdef decode_from_stream(self, InputStream in_stream, size_t length):
- return LocalTimeCoderImpl._decode_from_stream(in_stream)
+ return self._decode_from_stream(in_stream)
cdef class LocalDateTimeCoderImpl(FieldCoderImpl):
diff --git a/flink-python/pyflink/table/tests/test_table_environment_api.py
b/flink-python/pyflink/table/tests/test_table_environment_api.py
index bff9d89f3c8..1d22174e0b6 100644
--- a/flink-python/pyflink/table/tests/test_table_environment_api.py
+++ b/flink-python/pyflink/table/tests/test_table_environment_api.py
@@ -351,6 +351,23 @@ class DataStreamConversionTestCases(PyFlinkUTTestCase):
expected = ['+I[1]', '+I[2]', '+I[3]']
self.assertEqual(expected, results)
+ def test_to_data_stream_local_time(self):
+ self.t_env.execute_sql("""
+ CREATE TEMPORARY VIEW v0 AS
+ SELECT f0, f1, f2, f3 FROM ( VALUES
+ ( 1, DATE'1970-01-02', TIME'03:04:05', TIMESTAMP'1970-01-02
03:04:05' ),
+ ( 2, DATE'1970-06-07', TIME'08:09:10', TIMESTAMP'1970-06-07
08:09:10' )
+ ) AS t0 ( f0, f1, f2, f3 )
+ """)
+ v0 = self.t_env.from_path("v0")
+ self.t_env.to_data_stream(v0).key_by(lambda r:
r['f0']).add_sink(self.test_sink)
+ self.env.execute()
+ results = self.test_sink.get_results(False)
+ results.sort()
+ expected = ['+I[1, 1970-01-02, 03:04:05, 1970-01-02T03:04:05]',
+ '+I[2, 1970-06-07, 08:09:10, 1970-06-07T08:09:10]']
+ self.assertEqual(expected, results)
+
def test_from_data_stream(self):
self.env.set_parallelism(1)
