[FLINK-18763][python] Support basic TypeInformation for Python DataSt… #13029

shuiqiangchen · 2020-07-30T12:07:23Z

What is the purpose of the change

Supports basic TypeInformation including BasicTypeInfo, LocalTimeTypeInfo, PrimitiveArrayTypeInfo, RowTypeInfo.

Types.ROW()/Types.ROW_NAMED()/Types.PRIMITIVE_ARRAY() should also be supported.

Brief change log

Add python wrapper class for BasicTypeInformation, PrimitiveArrayTypeInfos and RowTypeInfo, etc.
Introduce a new type info named PickledArrayTypeInfo to indicate the type of bytes array data derived from pickled python objects.
Add utils to convert python typeInfo and java typeInfo.

Verifying this change

This change is already covered by existing tests, such as:

TypeInfoTests in test_typeinfo.py

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): ( no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? ( not documented)

…ream API.

flinkbot · 2020-07-30T12:09:44Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 2cb0d81 (Thu Jul 30 12:09:44 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-07-30T12:24:29Z

CI report:

1a3c5b9 UNKNOWN
2c7ff6c Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

hequn8128

@shuiqiangchen Thanks a lot for your PR. Left some comments below.

hequn8128 · 2020-07-31T02:57:01Z

flink-python/pyflink/common/typeinfo.py

+    TypeInformation is the core class of Flink's type system. FLink requires a type information
+    for all types that are used as input or return type of a user function. This type information
+    class acts as the tool to generate serializers and comparators, and to perform semantic checks
+    such as whether the fields that are used as join/grouping keys actually existt.


existt => exist

Thank you! I will revise it.

hequn8128 · 2020-07-31T03:00:51Z

flink-python/pyflink/common/typeinfo.py

+    in the arrays.
+        a) Basic types are indivisible and are considered as a single field.
+        b) Arrays and collections are one field.
+        c) Tuples and case classes represent as many fields as the class has fields.


We don't have case classes in PyFlink.

Yes, I will remove it.

hequn8128 · 2020-07-31T03:06:23Z

flink-python/pyflink/common/typeinfo.py

+    """
+
+    @abstractmethod
+    def is_basic_type(self):


Add type hint: def is_basic_type(self) -> bool:. Same for other methods.

Ok, it's good to add type hints.

hequn8128 · 2020-07-31T03:08:23Z

flink-python/pyflink/common/typeinfo.py

+    def is_tuple_type(self):
+        """
+        Checks if this type information represents a Tuple type.
+        Tuple types are subclasses of the Java API tuples.


We can remove this line. Tuple type is supported in Python originally. We should not simply copy Java comments.

Sure, there should be differences between java and python comments.

hequn8128 · 2020-07-31T03:12:22Z

flink-python/pyflink/common/typeinfo.py

+        nested fields, in the case of composite types.
+        The total number of fields must be at lest 1.
+
+        :return: The number of fields in this type, including its sub-fields (for composit types).


composite types

Thank you! I'll revise it.

hequn8128 · 2020-07-31T04:04:22Z

flink-python/pyflink/common/typeinfo.py

+                               .PickledByteArrayTypeInfo())
+
+
+class BigDecimalTypeInfo(WrapperTypeInfo):


Decimal should exist in BasicTypeInfo together with BIG_INT_TYPE_INFO. The decimal type info in DataStream is different from table. The precision and scale are not fixed, i.e., not be specified during initialization.

Yes, I'll remove it.

flink-python/pyflink/common/typeinfo.py

hequn8128 · 2020-07-31T07:23:04Z

...thon/src/main/java/org/apache/flink/datastream/typeinfo/python/PickledByteArrayTypeInfo.java

+ * A PickledByteArrayTypeInfo indicates that the data of this type is a generated primitive byte
+ * array by pickle.
+ */
+public class PickledByteArrayTypeInfo extends TypeInformation<byte[]> {


Make it a Singleton, similar to BasicTypeInfo in which provides a static final instance.

Ok, making it singleton is better.

…unnecessary types.

shuiqiangchen · 2020-07-31T08:00:18Z

Hi @hequn8128 , thank you for your comments, I have updated the pr according to your review suggestions, please have a look at it.

hequn8128

More comments below

hequn8128 · 2020-07-31T08:07:10Z

flink-python/pyflink/common/typeinfo.py

+        field_names = [name for name in j_field_names]
+        return field_names
+
+    def get_field_index(self, field_name) -> int:


Add type hint.

hequn8128 · 2020-07-31T08:21:02Z

flink-python/pyflink/common/typeinfo.py

+            raise TypeError("Invalid element type for a primitive array.")
+
+
+def from_java_type(j_type_info: JavaObject) -> TypeInformation:


Private method. Rename to _from_java_type.

hequn8128 · 2020-07-31T08:21:20Z

flink-python/pyflink/common/typeinfo.py

+    raise TypeError("The java type info: %s is not supported in PyFlink currently." % j_type_info)
+
+
+def is_instance_of(java_object: JavaObject, java_type: Union[JavaObject, JavaClass]) -> bool:


Private method. Rename to _is_instance_of.

hequn8128 · 2020-07-31T08:21:49Z

flink-python/pyflink/common/typeinfo.py

+    PICKLED_BYTE_ARRAY = PickledBytesTypeInfo.PICKLED_BYTE_ARRAY_TYPE_INFO
+
+    @staticmethod
+    def ROW(types):


Add type hint

hequn8128 · 2020-07-31T08:21:56Z

flink-python/pyflink/common/typeinfo.py

+        return RowTypeInfo(types)
+
+    @staticmethod
+    def ROW_NAMED(names, types):


Add type hint

flink-python/pyflink/common/typeinfo.py

hequn8128 · 2020-07-31T08:28:37Z

flink-python/pyflink/common/tests/test_typeinfo.py

+from pyflink.common.typeinfo import Types, RowTypeInfo
+
+
+class TypeInfoTests(unittest.TestCase):


Add tests to cover TupleTypeInfo and methods in Types(Types.ROW(), Types.ROW_NAMED, etc).

...thon/src/main/java/org/apache/flink/datastream/typeinfo/python/PickledByteArrayTypeInfo.java

hequn8128

@shuiqiangchen Thanks a lot for the update. The code looks good overall. Some minor comments below.

hequn8128 · 2020-08-03T01:57:49Z

flink-python/pyflink/common/tests/test_typeinfo.py

+                         TupleTypeInfo([Types.STRING(), Types.INT()]), True)
+
+        self.assertEqual(TupleTypeInfo([Types.STRING(), Types.INT()]).__str__(),
+                         "Java Tuple2<String, Integer>")


It is strange to return a string with Java for Python typeinfo. We need to regenerate the string for TupleTypeinfo.

Yes, maybe we should not directly call to toString() of the java type info object.

hequn8128 · 2020-08-03T01:59:36Z

flink-python/pyflink/common/tests/test_typeinfo.py

+
+
+


Remove the useless empty lines.

Thanks, I'll clean it.

hequn8128 · 2020-08-03T03:31:03Z

flink-python/pyflink/common/tests/test_typeinfo.py

+                                                                ['a', 'b']), False)
+        self.assertEqual(RowTypeInfo([Types.STRING(),
+                                      Types.STRING()],
+                                     ['a', 'b']).__str__(), "RowTypeInfo[String, String]")


We should also print field name for row type info.

hequn8128

Looks good from my side. Will merge this if @twalthr does not have any other concerns.

Best,
Hequn

hequn8128

@shuiqiangchen My last comments. Maybe we can remove the methods in the python TypeInformation since we don't need them now. It is always easier to add APIs than delete them. We can add these methods later when we find it is useful.

hequn8128 · 2020-08-04T09:09:48Z

flink-python/pyflink/common/typeinfo.py

+    def is_basic_type(self) -> bool:
+        """
+        Checks if this type information represents a basic type.
+        Basic types are defined in BasicTypeInfo and are primitives, their boxing type, Strings ...
+
+        :return:  True, if this type information describes a basic type, false otherwise.
+        """
+        pass
+
+    @abstractmethod
+    def is_tuple_type(self) -> bool:
+        """
+        Checks if this type information represents a Tuple type.
+
+        :return: True, if this type information describes a tuple type, false otherwise.
+        """
+        pass
+
+    @abstractmethod
+    def get_arity(self) -> int:
+        """
+        Gets the arity of this type - the number of fields without nesting.
+
+        :return: the number of fields in this type without nesting.
+        """
+        pass
+
+    @abstractmethod
+    def get_total_fields(self) -> int:
+        """
+        Gets the number of logical fields in this type. This includes its nested and transitively
+        nested fields, in the case of composite types.
+        The total number of fields must be at lest 1.
+
+        :return: The number of fields in this type, including its sub-fields (for composite types).
+        """
+        pass


We can remove these methods since we don't need them now.

Agree, I will remove them.

hequn8128 · 2020-08-04T09:10:10Z

flink-python/pyflink/common/typeinfo.py

+    def is_basic_type(self) -> bool:
+        return self._j_typeinfo.isBasicType()
+
+    def is_tuple_type(self) -> bool:
+        return self._j_typeinfo.isTupleType()
+
+    def get_arity(self) -> int:
+        return self._j_typeinfo.getArity()
+
+    def get_total_fields(self) -> int:


shuiqiangchen · 2020-08-04T09:31:03Z

@hequn8128 Thank you for your comments, I have updated the code, please take a look at it.

hequn8128 · 2020-08-04T12:01:24Z

@shuiqiangchen Thanks a lot for the update. Please check the azure test failures.

[FLINK-18763][python] Support basic TypeInformation for Python DataSt…

2cb0d81

…ream API.

rmetzger added the review=description? label Jul 30, 2020

rmetzger added the component=<none> label Jul 30, 2020

shuiqiangchen added 3 commits July 31, 2020 10:07

- remove BasicTypeInfo.VOID and BasicTypeInfo.DATE.

fd04ce4

- resolve checkstyle error.

3d89fc0

- remove BigDecimalTypeInfo, use BasicTypeInfo.BIG_DEC instead.

39f5be6

hequn8128 requested changes Jul 31, 2020

View reviewed changes

shuiqiangchen added 2 commits July 31, 2020 15:25

- remove LocalDateTimeTypeInfo.

1a3c5b9

- revise wrong words in comments, add type hints for methods, remove …

c583718

…unnecessary types.

- remove INSTANT typeinfo and correct comments.

9e004f9

hequn8128 requested changes Jul 31, 2020

View reviewed changes

- add type hint for methods, add more tests for typeinfos.

2d8d302

hequn8128 requested changes Aug 3, 2020

View reviewed changes

shuiqiangchen added 2 commits August 3, 2020 10:22

- refactor code style.

6f2353a

- refactor __str()__ for type infos, optimize test cases.

dc01ebc

hequn8128 reviewed Aug 3, 2020

View reviewed changes

- print field name when calling RowTypeInfo.__str()__.

be7e94a

hequn8128 reviewed Aug 3, 2020

View reviewed changes

hequn8128 requested changes Aug 4, 2020

View reviewed changes

- remove unnecessary methods in TypeInformation.

943ff7e

- remove is_basic_type_test.

2c7ff6c

hequn8128 merged commit ea9f449 into apache:master Aug 5, 2020

flinkbot added component=API/Python and removed component=<none> labels Mar 17, 2022

		.PickledByteArrayTypeInfo())


		class BigDecimalTypeInfo(WrapperTypeInfo):

		raise TypeError("Invalid element type for a primitive array.")


		def from_java_type(j_type_info: JavaObject) -> TypeInformation:

		raise TypeError("The java type info: %s is not supported in PyFlink currently." % j_type_info)


		def is_instance_of(java_object: JavaObject, java_type: Union[JavaObject, JavaClass]) -> bool:

		from pyflink.common.typeinfo import Types, RowTypeInfo


		class TypeInfoTests(unittest.TestCase):

[FLINK-18763][python] Support basic TypeInformation for Python DataSt… #13029

[FLINK-18763][python] Support basic TypeInformation for Python DataSt… #13029

Uh oh!

Conversation

shuiqiangchen commented Jul 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jul 30, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Jul 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

hequn8128 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shuiqiangchen commented Jul 31, 2020

Uh oh!

hequn8128 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

hequn8128 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shuiqiangchen commented Jul 30, 2020 •

edited

Loading

flinkbot commented Jul 30, 2020 •

edited

Loading

hequn8128 commented Aug 4, 2020 •

edited

Loading