You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2022/08/15 04:36:41 UTC
[flink] branch release-1.15 updated: [FLINK-28908][python] Fix LIST type in Python DataStream API
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 63c1c1b9f6e [FLINK-28908][python] Fix LIST type in Python DataStream API
63c1c1b9f6e is described below
commit 63c1c1b9f6ee9669305e081030768b43a267c0f5
Author: Juntao Hu <ma...@gmail.com>
AuthorDate: Thu Aug 11 10:19:38 2022 +0800
[FLINK-28908][python] Fix LIST type in Python DataStream API
This closes #20539.
---
flink-python/pyflink/common/typeinfo.py | 4 ++--
flink-python/pyflink/datastream/tests/test_data_stream.py | 8 ++++++++
flink-python/pyflink/fn_execution/coders.py | 12 ++++++++----
flink-python/pyflink/table/types.py | 2 +-
4 files changed, 19 insertions(+), 7 deletions(-)
diff --git a/flink-python/pyflink/common/typeinfo.py b/flink-python/pyflink/common/typeinfo.py
index 5558fad683e..c1ece76f9d9 100644
--- a/flink-python/pyflink/common/typeinfo.py
+++ b/flink-python/pyflink/common/typeinfo.py
@@ -439,7 +439,7 @@ class RowTypeInfo(TypeInformation):
zip(self.get_field_names(), self._field_types, self._need_conversion))
elif isinstance(obj, Row) and hasattr(obj, "_fields"):
return (obj.get_row_kind().value,) + tuple(
- f.to_internal_type(obj.get(n)) if c else obj.get(n)
+ f.to_internal_type(obj[n]) if c else obj[n]
for n, f, c in
zip(self.get_field_names(), self._field_types, self._need_conversion))
elif isinstance(obj, Row):
@@ -463,7 +463,7 @@ class RowTypeInfo(TypeInformation):
return (RowKind.INSERT.value,) + tuple(obj.get(n) for n in self.get_field_names())
elif isinstance(obj, Row) and hasattr(obj, "_fields"):
return (obj.get_row_kind().value,) + tuple(
- obj.get(n) for n in self.get_field_names())
+ obj[n] for n in self.get_field_names())
elif isinstance(obj, Row):
return (obj.get_row_kind().value,) + tuple(obj)
elif isinstance(obj, (list, tuple)):
diff --git a/flink-python/pyflink/datastream/tests/test_data_stream.py b/flink-python/pyflink/datastream/tests/test_data_stream.py
index aa15dc21af7..3fd0a9fb372 100644
--- a/flink-python/pyflink/datastream/tests/test_data_stream.py
+++ b/flink-python/pyflink/datastream/tests/test_data_stream.py
@@ -812,6 +812,14 @@ class DataStreamTests(object):
expected = ['(hi,3)']
self.assert_equals_sorted(expected, results)
+ def test_java_list_deserialization(self):
+ row_type_info = Types.ROW_NAMED(['list'], [Types.LIST(Types.INT())])
+ ds = self.env.from_collection([Row(list=[1, 2, 3])], type_info=row_type_info)
+ ds.map(lambda e: str(e), Types.STRING()).add_sink(self.test_sink)
+ self.env.execute('test_java_list_deserialization')
+ expected = ['Row(list=[1, 2, 3])']
+ self.assert_equals(self.test_sink.get_results(), expected)
+
class StreamingModeDataStreamTests(DataStreamTests, PyFlinkStreamingTestCase):
def test_data_stream_name(self):
diff --git a/flink-python/pyflink/fn_execution/coders.py b/flink-python/pyflink/fn_execution/coders.py
index 7e43f78b2b2..3f3da97535f 100644
--- a/flink-python/pyflink/fn_execution/coders.py
+++ b/flink-python/pyflink/fn_execution/coders.py
@@ -679,13 +679,17 @@ def from_type_info_proto(type_info):
return RowCoder(
[from_type_info_proto(f.field_type) for f in type_info.row_type_info.fields],
[f.field_name for f in type_info.row_type_info.fields])
- elif field_type_name == type_info_name.PRIMITIVE_ARRAY:
+ elif field_type_name in (
+ type_info_name.PRIMITIVE_ARRAY,
+ type_info_name.LIST,
+ ):
if type_info.collection_element_type.type_name == type_info_name.BYTE:
return BinaryCoder()
return PrimitiveArrayCoder(from_type_info_proto(type_info.collection_element_type))
- elif field_type_name in (type_info_name.BASIC_ARRAY,
- type_info_name.OBJECT_ARRAY,
- type_info_name.LIST):
+ elif field_type_name in (
+ type_info_name.BASIC_ARRAY,
+ type_info_name.OBJECT_ARRAY,
+ ):
return GenericArrayCoder(from_type_info_proto(type_info.collection_element_type))
elif field_type_name == type_info_name.TUPLE:
return TupleCoder([from_type_info_proto(field_type)
diff --git a/flink-python/pyflink/table/types.py b/flink-python/pyflink/table/types.py
index 44458da9b3c..8217844b013 100644
--- a/flink-python/pyflink/table/types.py
+++ b/flink-python/pyflink/table/types.py
@@ -1250,7 +1250,7 @@ class RowType(DataType):
for n, f, c in zip(self.names, self.fields, self._need_conversion))
elif isinstance(obj, Row) and hasattr(obj, "_fields"):
return (obj.get_row_kind().value,) + tuple(
- f.to_sql_type(obj.get(n)) if c else obj.get(n)
+ f.to_sql_type(obj[n]) if c else obj[n]
for n, f, c in zip(self.names, self.fields, self._need_conversion))
elif isinstance(obj, Row):
return (obj.get_row_kind().value, ) + tuple(