[FLINK-22612][python] Restructure the coders in PyFlink #15877

HuangXingBo · 2021-05-10T08:20:53Z

What is the purpose of the change

This pull request will restructure the coders in PyFlink

Brief change log

After restructuring the coders in PyFlink, there will be only two top-level coders including B, IterableCoder and ValueCoder

Verifying this change

Original Tests

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)
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? (no)
If yes, how is the feature documented? (not applicable)

flinkbot · 2021-05-10T08:38:27Z

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 f607214 (Sat Aug 28 12:14:45 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{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.

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 · 2021-05-10T09:03:03Z

CI report:

f607214 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

dianfu · 2021-05-13T12:03:43Z

flink-python/pyflink/fn_execution/beam/beam_coders.py

-    """
-    Coder for Table Function Row.
-    """
+class BeamCoder(FastCoder):


The name BeamCoder seems not quite suitable. What about rename it to FlinkCoder?

dianfu · 2021-05-13T12:10:38Z

flink-python/pyflink/fn_execution/coders.py

+            schema_proto = coder_param_proto.schema
+            row_type = cls._to_row_type(schema_proto)
+            return ArrowCoder(cls._to_arrow_schema(row_type), row_type, timezone, output_mode)
+        elif data_type == flink_fn_execution_pb2.CoderParam.BATCH_OVER_WINDOW_ARROW:


What about rename it to OVER_WINDOW_ARROW to match with the class name OverWindowArrowCoder ?

dianfu · 2021-05-14T02:20:11Z

flink-python/pyflink/fn_execution/beam/beam_coders.py

-    FloatType, DoubleType, VarCharType, VarBinaryType, DecimalType, DateType, TimeType, \
-    LocalZonedTimestampType, RowType, RowField, to_arrow_type, TimestampType, ArrayType
+
+FLINK_CODER_URN = "flink:coder"


Suggested change

FLINK_CODER_URN = "flink:coder"

FLINK_CODER_URN = "flink:coder:v1"

dianfu · 2021-05-31T06:35:55Z

flink-python/pyflink/fn_execution/beam/beam_coder_impl_slow.py

 from pyflink.table.utils import pandas_to_arrow, arrow_to_pandas

 ROW_KIND_BIT_SIZE = 2


-class TimeWindowCoderImpl(StreamCoderImpl):
+class BaseCoderImpl(ABC, StreamCoderImpl):


Suggested change

class BaseCoderImpl(ABC, StreamCoderImpl):

class LengthPrefixBaseCoderImpl(ABC, StreamCoderImpl):

dianfu · 2021-05-31T07:36:42Z

flink-python/pyflink/fn_execution/operation_utils.py

@@ -44,6 +45,34 @@
 _constant_num = 0


+def wrap_table_function_result(it):


Suggested change

def wrap_table_function_result(it):

def normalize_table_function_result(it):

rmetzger added the review=description? label May 10, 2021

HuangXingBo force-pushed the FLINK-22612 branch from acd4db6 to ae350a4 Compare May 10, 2021 09:06

rmetzger added the component=API/Python label May 10, 2021

HuangXingBo force-pushed the FLINK-22612 branch 2 times, most recently from 816de48 to 7ba1fe4 Compare May 11, 2021 08:32

dianfu reviewed May 14, 2021

View reviewed changes

HuangXingBo force-pushed the FLINK-22612 branch 2 times, most recently from dc67929 to 12d8b6d Compare May 25, 2021 11:34

HuangXingBo added 4 commits May 31, 2021 10:06

[FLINK-22612][python] Restructure the coders in PyFlink

bfc76f2

fix comments

6894a1f

fix

68e271a

fix datastream test

c37690c

HuangXingBo force-pushed the FLINK-22612 branch from 12d8b6d to c37690c Compare May 31, 2021 02:06

dianfu reviewed May 31, 2021

View reviewed changes

fix comments

63faa9b

HuangXingBo force-pushed the FLINK-22612 branch from d92da78 to 63faa9b Compare June 2, 2021 05:11

fix test

f607214

dianfu approved these changes Jun 3, 2021

View reviewed changes

HuangXingBo closed this in 9986e6f Jun 3, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-22612][python] Restructure the coders in PyFlink #15877

[FLINK-22612][python] Restructure the coders in PyFlink #15877

HuangXingBo commented May 10, 2021 •

edited

flinkbot commented May 10, 2021 •

edited

flinkbot commented May 10, 2021 •

edited

dianfu May 13, 2021

dianfu May 13, 2021

dianfu May 14, 2021

dianfu May 31, 2021

dianfu May 31, 2021

	FLINK_CODER_URN = "flink:coder"
	FLINK_CODER_URN = "flink:coder:v1"

	class BaseCoderImpl(ABC, StreamCoderImpl):
	class LengthPrefixBaseCoderImpl(ABC, StreamCoderImpl):

		@@ -44,6 +45,34 @@
		_constant_num = 0


		def wrap_table_function_result(it):

	def wrap_table_function_result(it):
	def normalize_table_function_result(it):

[FLINK-22612][python] Restructure the coders in PyFlink #15877

[FLINK-22612][python] Restructure the coders in PyFlink #15877

Conversation

HuangXingBo commented May 10, 2021 • edited

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

flinkbot commented May 10, 2021 • edited

Automated Checks

Review Progress

flinkbot commented May 10, 2021 • edited

CI report:

dianfu May 13, 2021

Choose a reason for hiding this comment

dianfu May 13, 2021

Choose a reason for hiding this comment

dianfu May 14, 2021

Choose a reason for hiding this comment

dianfu May 31, 2021

Choose a reason for hiding this comment

dianfu May 31, 2021

Choose a reason for hiding this comment

HuangXingBo commented May 10, 2021 •

edited

flinkbot commented May 10, 2021 •

edited

flinkbot commented May 10, 2021 •

edited