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 2020/04/19 13:31:20 UTC
[flink] branch master updated (e0d2db7 -> 3cd8716)
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from e0d2db7 [FLINK-16366] [table] Introduce executeSql method in TableEnvironment
new c874700 [FLINK-17118][AZP] Use rmetzger/flink-ci:ubuntu-amd64-bcef226 image which contains gcc used for PyFlink cython tests
new 3cd8716 [FLINK-17118][python] Add Cython support for primitive data types
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.gitignore | 2 +
azure-pipelines.yml | 2 +-
flink-python/MANIFEST.in | 2 +
flink-python/pyflink/fn_execution/coder_impl.py | 2 +-
flink-python/pyflink/fn_execution/coders.py | 2 +-
.../pyflink/fn_execution/fast_coder_impl.pxd | 192 +++++++
.../pyflink/fn_execution/fast_coder_impl.pyx | 553 +++++++++++++++++++++
.../{test_coders_common.py => test_coders.py} | 9 +
.../pyflink/fn_execution/tests/test_fast_coders.py | 141 ++++++
flink-python/setup.py | 28 +-
flink-python/tox.ini | 8 +-
tools/azure-pipelines/build-apache-repo.yml | 2 +-
12 files changed, 936 insertions(+), 7 deletions(-)
create mode 100644 flink-python/pyflink/fn_execution/fast_coder_impl.pxd
create mode 100644 flink-python/pyflink/fn_execution/fast_coder_impl.pyx
rename flink-python/pyflink/fn_execution/tests/{test_coders_common.py => test_coders.py} (95%)
create mode 100644 flink-python/pyflink/fn_execution/tests/test_fast_coders.py
[flink] 01/02: [FLINK-17118][AZP] Use
rmetzger/flink-ci:ubuntu-amd64-bcef226 image which contains gcc used for
PyFlink cython tests
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c874700d35fc28cb9af4fc4255c54afa3aaf455e
Author: huangxingbo <hx...@gmail.com>
AuthorDate: Sun Apr 19 17:32:09 2020 +0800
[FLINK-17118][AZP] Use rmetzger/flink-ci:ubuntu-amd64-bcef226 image which contains gcc used for PyFlink cython tests
---
azure-pipelines.yml | 2 +-
tools/azure-pipelines/build-apache-repo.yml | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index e0a5853..d1966a9 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -38,7 +38,7 @@ resources:
containers:
# Container with Maven 3.2.5, SSL to have the same environment everywhere.
- container: flink-build-container
- image: rmetzger/flink-ci:ubuntu-amd64-3528acd
+ image: rmetzger/flink-ci:ubuntu-amd64-bcef226
# On AZP provided machines, set this flag to allow writing coredumps in docker
options: --privileged
diff --git a/tools/azure-pipelines/build-apache-repo.yml b/tools/azure-pipelines/build-apache-repo.yml
index 1906fca..96979e0 100644
--- a/tools/azure-pipelines/build-apache-repo.yml
+++ b/tools/azure-pipelines/build-apache-repo.yml
@@ -38,7 +38,7 @@ resources:
containers:
# Container with Maven 3.2.5, SSL to have the same environment everywhere.
- container: flink-build-container
- image: rmetzger/flink-ci:ubuntu-amd64-3528acd
+ image: rmetzger/flink-ci:ubuntu-amd64-bcef226
variables:
MAVEN_CACHE_FOLDER: $(Pipeline.Workspace)/.m2/repository
[flink] 02/02: [FLINK-17118][python] Add Cython support for
primitive data types
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3cd87166b7f134a520a842f917376043ffb9cbb1
Author: huangxingbo <hx...@gmail.com>
AuthorDate: Sun Apr 19 17:34:23 2020 +0800
[FLINK-17118][python] Add Cython support for primitive data types
This closes #11809.
---
.gitignore | 2 +
flink-python/MANIFEST.in | 2 +
flink-python/pyflink/fn_execution/coder_impl.py | 2 +-
flink-python/pyflink/fn_execution/coders.py | 2 +-
.../pyflink/fn_execution/fast_coder_impl.pxd | 192 +++++++
.../pyflink/fn_execution/fast_coder_impl.pyx | 553 +++++++++++++++++++++
.../{test_coders_common.py => test_coders.py} | 9 +
.../pyflink/fn_execution/tests/test_fast_coders.py | 141 ++++++
flink-python/setup.py | 28 +-
flink-python/tox.ini | 8 +-
10 files changed, 934 insertions(+), 5 deletions(-)
diff --git a/.gitignore b/.gitignore
index e89f622..886d61b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -36,6 +36,8 @@ flink-python/dev/.conda/
flink-python/dev/log/
flink-python/dev/.stage.txt
flink-python/.eggs/
+flink-python/**/*.c
+flink-python/**/*.so
atlassian-ide-plugin.xml
out/
/docs/api
diff --git a/flink-python/MANIFEST.in b/flink-python/MANIFEST.in
index a8e2d61..edef99d 100644
--- a/flink-python/MANIFEST.in
+++ b/flink-python/MANIFEST.in
@@ -30,3 +30,5 @@ include pyflink/LICENSE
include pyflink/NOTICE
include pyflink/README.txt
recursive-exclude deps/opt/python *
+recursive-include pyflink/fn_execution *.pxd
+recursive-include pyflink/fn_execution *.pyx
diff --git a/flink-python/pyflink/fn_execution/coder_impl.py b/flink-python/pyflink/fn_execution/coder_impl.py
index 743ce7b..d4aac4d 100644
--- a/flink-python/pyflink/fn_execution/coder_impl.py
+++ b/flink-python/pyflink/fn_execution/coder_impl.py
@@ -237,7 +237,7 @@ class TinyIntCoderImpl(StreamCoderImpl):
return struct.unpack('b', in_stream.read(1))[0]
-class SmallIntImpl(StreamCoderImpl):
+class SmallIntCoderImpl(StreamCoderImpl):
def encode_to_stream(self, value, out_stream, nested):
out_stream.write(struct.pack('>h', value))
diff --git a/flink-python/pyflink/fn_execution/coders.py b/flink-python/pyflink/fn_execution/coders.py
index d2be4fd..4d9bfda 100644
--- a/flink-python/pyflink/fn_execution/coders.py
+++ b/flink-python/pyflink/fn_execution/coders.py
@@ -262,7 +262,7 @@ class SmallIntCoder(DeterministicCoder):
"""
def _create_impl(self):
- return coder_impl.SmallIntImpl()
+ return coder_impl.SmallIntCoderImpl()
def to_type_hint(self):
return int
diff --git a/flink-python/pyflink/fn_execution/fast_coder_impl.pxd b/flink-python/pyflink/fn_execution/fast_coder_impl.pxd
new file mode 100644
index 0000000..2156247
--- /dev/null
+++ b/flink-python/pyflink/fn_execution/fast_coder_impl.pxd
@@ -0,0 +1,192 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+# cython: language_level=3
+
+cimport libc.stdint
+
+from apache_beam.coders.coder_impl cimport StreamCoderImpl, OutputStream, InputStream
+
+# InputStreamAndFunctionWrapper wraps the user-defined function
+# and input_stream_wrapper in operations
+cdef class InputStreamAndFunctionWrapper:
+ # user-defined function
+ cdef readonly object func
+ cdef InputStreamWrapper input_stream_wrapper
+
+# InputStreamWrapper wraps input_stream and related infos used to decode data
+cdef class InputStreamWrapper:
+ cdef InputStream input_stream
+ cdef list input_field_coders
+ cdef TypeName*input_field_type
+ cdef CoderType*input_coder_type
+ cdef libc.stdint.int32_t input_field_count
+ cdef libc.stdint.int32_t input_leading_complete_bytes_num
+ cdef libc.stdint.int32_t input_remaining_bits_num
+ cdef size_t input_buffer_size
+
+cdef class PassThroughLengthPrefixCoderImpl(StreamCoderImpl):
+ cdef readonly StreamCoderImpl _value_coder
+
+cdef class FlattenRowCoderImpl(StreamCoderImpl):
+ # the input field coders and related args used to decode input_stream data
+ cdef list _input_field_coders
+ cdef TypeName*_input_field_type
+ cdef CoderType*_input_coder_type
+ cdef libc.stdint.int32_t _input_field_count
+ cdef libc.stdint.int32_t _input_leading_complete_bytes_num
+ cdef libc.stdint.int32_t _input_remaining_bits_num
+
+ # the output field coders and related args used to encode data to output_stream
+ cdef readonly list _output_field_coders
+ cdef TypeName*_output_field_type
+ cdef CoderType*_output_coder_type
+ cdef libc.stdint.int32_t _output_field_count
+ cdef libc.stdint.int32_t _output_leading_complete_bytes_num
+ cdef libc.stdint.int32_t _output_remaining_bits_num
+
+ cdef bint*_null_mask
+ cdef unsigned char*_null_byte_search_table
+
+ # the char pointer used to store encoded data of output_stream
+ cdef char*_output_data
+ cdef size_t _output_buffer_size
+ cdef size_t _output_pos
+
+ # the tmp char pointer used to store encoded data of every row
+ cdef char*_tmp_output_data
+ cdef size_t _tmp_output_buffer_size
+ cdef size_t _tmp_output_pos
+
+ # the char pointer used to map the decoded data of input_stream
+ cdef char*_input_data
+ cdef size_t _input_pos
+ cdef size_t _input_buffer_size
+
+ # used to store the result of Python user-defined function
+ cdef list row
+
+ # the Python user-defined function
+ cdef object func
+
+ # initial attribute
+ cdef void _init_attribute(self)
+
+ # wrap input_stream
+ cdef InputStreamWrapper _wrap_input_stream(self, InputStream input_stream, size_t size)
+
+ cdef void _write_null_mask(self, value, libc.stdint.int32_t leading_complete_bytes_num,
+ libc.stdint.int32_t remaining_bits_num)
+ cdef void _read_null_mask(self, bint*null_mask, libc.stdint.int32_t leading_complete_bytes_num,
+ libc.stdint.int32_t remaining_bits_num)
+
+ cdef void _prepare_encode(self, InputStreamAndFunctionWrapper input_stream_and_function_wrapper,
+ OutputStream out_stream)
+
+ cdef void _maybe_flush(self, OutputStream out_stream)
+ # Because output_buffer will be reallocated during encoding data, we need to remap output_buffer
+ # to the data pointer of output_stream
+ cdef void _map_output_data_to_output_stream(self, OutputStream out_stream)
+ cdef void _copy_to_output_buffer(self)
+
+ # encode data to output_stream
+ cdef void _encode_one_row(self, value)
+ cdef void _encode_field_simple(self, TypeName field_type, item)
+ cdef void _extend(self, size_t missing)
+ cdef void _encode_byte(self, unsigned char val)
+ cdef void _encode_smallint(self, libc.stdint.int16_t v)
+ cdef void _encode_int(self, libc.stdint.int32_t v)
+ cdef void _encode_bigint(self, libc.stdint.int64_t v)
+ cdef void _encode_float(self, float v)
+ cdef void _encode_double(self, double v)
+ cdef void _encode_bytes(self, char*b)
+
+ # decode data from input_stream
+ cdef void _decode_next_row(self)
+ cdef object _decode_field_simple(self, TypeName field_type)
+ cdef unsigned char _decode_byte(self) except? -1
+ cdef libc.stdint.int16_t _decode_smallint(self) except? -1
+ cdef libc.stdint.int32_t _decode_int(self) except? -1
+ cdef libc.stdint.int64_t _decode_bigint(self) except? -1
+ cdef float _decode_float(self) except? -1
+ cdef double _decode_double(self) except? -1
+ cdef bytes _decode_bytes(self)
+
+cdef class TableFunctionRowCoderImpl(FlattenRowCoderImpl):
+ cdef void _encode_end_message(self)
+
+cdef enum CoderType:
+ UNDEFINED = -1
+ SIMPLE = 0
+ COMPLEX = 1
+
+cdef enum TypeName:
+ NONE = -1
+ ROW = 0
+ TINYINT = 1
+ SMALLINT = 2
+ INT = 3
+ BIGINT = 4
+ DECIMAL = 5
+ FLOAT = 6
+ DOUBLE = 7
+ DATE = 8
+ TIME = 9
+ TIMESTAMP = 10
+ BOOLEAN = 11
+ BINARY = 12
+ CHAR = 13
+ ARRAY = 14
+ MAP = 15
+ LOCAL_ZONED_TIMESTAMP = 16
+
+cdef class BaseCoder:
+ cpdef CoderType coder_type(self)
+ cpdef TypeName type_name(self)
+
+cdef class TinyIntCoderImpl(BaseCoder):
+ pass
+
+cdef class SmallIntCoderImpl(BaseCoder):
+ pass
+
+cdef class IntCoderImpl(BaseCoder):
+ pass
+
+cdef class BigIntCoderImpl(BaseCoder):
+ pass
+
+cdef class BooleanCoderImpl(BaseCoder):
+ pass
+
+cdef class FloatCoderImpl(BaseCoder):
+ pass
+
+cdef class DoubleCoderImpl(BaseCoder):
+ pass
+
+cdef class BinaryCoderImpl(BaseCoder):
+ pass
+
+cdef class CharCoderImpl(BaseCoder):
+ pass
+
+cdef class DateCoderImpl(BaseCoder):
+ pass
+
+cdef class TimeCoderImpl(BaseCoder):
+ pass
diff --git a/flink-python/pyflink/fn_execution/fast_coder_impl.pyx b/flink-python/pyflink/fn_execution/fast_coder_impl.pyx
new file mode 100644
index 0000000..bfc08202
--- /dev/null
+++ b/flink-python/pyflink/fn_execution/fast_coder_impl.pyx
@@ -0,0 +1,553 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+# cython: language_level = 3
+# cython: infer_types = True
+# cython: profile=True
+# cython: boundscheck=False, wraparound=False, initializedcheck=False, cdivision=True
+
+cimport libc.stdlib
+from libc.string cimport strlen
+
+import datetime
+
+cdef class InputStreamAndFunctionWrapper:
+ def __cinit__(self, func, input_stream_wrapper):
+ self.func = func
+ self.input_stream_wrapper = input_stream_wrapper
+
+cdef class PassThroughLengthPrefixCoderImpl(StreamCoderImpl):
+ def __cinit__(self, value_coder):
+ self._value_coder = value_coder
+
+ cpdef encode_to_stream(self, value, OutputStream out_stream, bint nested):
+ self._value_coder.encode_to_stream(value, out_stream, nested)
+
+ cpdef decode_from_stream(self, InputStream in_stream, bint nested):
+ return self._value_coder.decode_from_stream(in_stream, nested)
+
+ cpdef get_estimated_size_and_observables(self, value, bint nested=False):
+ return 0, []
+
+cdef class TableFunctionRowCoderImpl(FlattenRowCoderImpl):
+ def __init__(self, flatten_row_coder):
+ super(TableFunctionRowCoderImpl, self).__init__(flatten_row_coder._output_field_coders)
+
+ cpdef encode_to_stream(self, input_stream_and_function_wrapper, OutputStream out_stream,
+ bint nested):
+ self._prepare_encode(input_stream_and_function_wrapper, out_stream)
+ while self._input_buffer_size > self._input_pos:
+ self._decode_next_row()
+ result = self.func(self.row)
+ if result:
+ for value in result:
+ if self._output_field_count == 1:
+ value = (value,)
+ self._encode_one_row(value)
+ self._maybe_flush(out_stream)
+ self._encode_end_message()
+
+ self._map_output_data_to_output_stream(out_stream)
+
+ # write 0x00 as end message
+ cdef void _encode_end_message(self):
+ if self._output_buffer_size < self._output_pos + 2:
+ self._extend(2)
+ self._output_data[self._output_pos] = 0x01
+ self._output_data[self._output_pos + 1] = 0x00
+ self._output_pos += 2
+
+cdef class FlattenRowCoderImpl(StreamCoderImpl):
+ def __init__(self, field_coders):
+ self._output_field_coders = field_coders
+ self._output_field_count = len(self._output_field_coders)
+ self._output_field_type = <TypeName*> libc.stdlib.malloc(
+ self._output_field_count * sizeof(TypeName))
+ self._output_coder_type = <CoderType*> libc.stdlib.malloc(
+ self._output_field_count * sizeof(CoderType))
+ self._output_leading_complete_bytes_num = self._output_field_count // 8
+ self._output_remaining_bits_num = self._output_field_count % 8
+ self._tmp_output_buffer_size = 1024
+ self._tmp_output_pos = 0
+ self._tmp_output_data = <char*> libc.stdlib.malloc(self._tmp_output_buffer_size)
+ self._null_byte_search_table = <unsigned char*> libc.stdlib.malloc(
+ 8 * sizeof(unsigned char))
+ self._init_attribute()
+
+ cpdef decode_from_stream(self, InputStream in_stream, bint nested):
+ cdef InputStreamWrapper input_stream_wrapper
+ input_stream_wrapper = self._wrap_input_stream(in_stream, in_stream.size())
+ return input_stream_wrapper
+
+ cpdef encode_to_stream(self, input_stream_and_function_wrapper, OutputStream out_stream,
+ bint nested):
+ cdef list result
+ self._prepare_encode(input_stream_and_function_wrapper, out_stream)
+ while self._input_buffer_size > self._input_pos:
+ self._decode_next_row()
+ result = self.func(self.row)
+ self._encode_one_row(result)
+ self._maybe_flush(out_stream)
+ self._map_output_data_to_output_stream(out_stream)
+
+ cdef void _init_attribute(self):
+ self._null_byte_search_table[0] = 0x80
+ self._null_byte_search_table[1] = 0x40
+ self._null_byte_search_table[2] = 0x20
+ self._null_byte_search_table[3] = 0x10
+ self._null_byte_search_table[4] = 0x08
+ self._null_byte_search_table[5] = 0x04
+ self._null_byte_search_table[6] = 0x02
+ self._null_byte_search_table[7] = 0x01
+ for i in range(self._output_field_count):
+ self._output_field_type[i] = self._output_field_coders[i].type_name()
+ self._output_coder_type[i] = self._output_field_coders[i].coder_type()
+
+ cdef InputStreamWrapper _wrap_input_stream(self, InputStream input_stream, size_t size):
+ # wrappers the input field coders and input_stream together
+ # so that it can be transposed to operations
+ cdef InputStreamWrapper input_stream_wrapper
+ input_stream_wrapper = InputStreamWrapper()
+ input_stream_wrapper.input_stream = input_stream
+ input_stream_wrapper.input_field_coders = self._output_field_coders
+ input_stream_wrapper.input_remaining_bits_num = self._output_remaining_bits_num
+ input_stream_wrapper.input_leading_complete_bytes_num = \
+ self._output_leading_complete_bytes_num
+ input_stream_wrapper.input_field_count = self._output_field_count
+ input_stream_wrapper.input_field_type = self._output_field_type
+ input_stream_wrapper.input_coder_type = self._output_coder_type
+ input_stream_wrapper.input_stream.pos = size
+ input_stream_wrapper.input_buffer_size = size
+ return input_stream_wrapper
+
+ cdef void _encode_one_row(self, value):
+ cdef libc.stdint.int32_t i
+ self._write_null_mask(value, self._output_leading_complete_bytes_num,
+ self._output_remaining_bits_num)
+ for i in range(self._output_field_count):
+ item = value[i]
+ if item is not None:
+ if self._output_coder_type[i] == SIMPLE:
+ self._encode_field_simple(self._output_field_type[i], item)
+
+ self._copy_to_output_buffer()
+
+ cdef void _read_null_mask(self, bint*null_mask,
+ libc.stdint.int32_t input_leading_complete_bytes_num,
+ libc.stdint.int32_t input_remaining_bits_num):
+ cdef libc.stdint.int32_t field_pos, i
+ cdef unsigned char b
+ field_pos = 0
+ for _ in range(input_leading_complete_bytes_num):
+ b = self._input_data[self._input_pos]
+ self._input_pos += 1
+ for i in range(8):
+ null_mask[field_pos] = (b & self._null_byte_search_table[i]) > 0
+ field_pos += 1
+
+ if input_remaining_bits_num:
+ b = self._input_data[self._input_pos]
+ self._input_pos += 1
+ for i in range(input_remaining_bits_num):
+ null_mask[field_pos] = (b & self._null_byte_search_table[i]) > 0
+ field_pos += 1
+
+ cdef void _decode_next_row(self):
+ cdef libc.stdint.int32_t i
+ # skip prefix variable int length
+ while self._input_data[self._input_pos] & 0x80:
+ self._input_pos += 1
+ self._input_pos += 1
+ self._read_null_mask(self._null_mask, self._input_leading_complete_bytes_num,
+ self._input_remaining_bits_num)
+ for i in range(self._input_field_count):
+ if self._null_mask[i]:
+ self.row[i] = None
+ else:
+ if self._input_coder_type[i] == SIMPLE:
+ self.row[i] = self._decode_field_simple(self._input_field_type[i])
+
+ cdef object _decode_field_simple(self, TypeName field_type):
+ cdef libc.stdint.int32_t value, minutes, seconds, hours
+ cdef libc.stdint.int64_t milliseconds
+ if field_type == TINYINT:
+ # tinyint
+ return self._decode_byte()
+ elif field_type == SMALLINT:
+ # smallint
+ return self._decode_smallint()
+ elif field_type == INT:
+ # int
+ return self._decode_int()
+ elif field_type == BIGINT:
+ # bigint
+ return self._decode_bigint()
+ elif field_type == BOOLEAN:
+ # boolean
+ return not not self._decode_byte()
+ elif field_type == FLOAT:
+ # float
+ return self._decode_float()
+ elif field_type == DOUBLE:
+ # double
+ return self._decode_double()
+ elif field_type == BINARY:
+ # bytes
+ return self._decode_bytes()
+ elif field_type == CHAR:
+ # str
+ return self._decode_bytes().decode("utf-8")
+ elif field_type == DATE:
+ # Date
+ # EPOCH_ORDINAL = datetime.datetime(1970, 1, 1).toordinal()
+ # The value of EPOCH_ORDINAL is 719163
+ return datetime.date.fromordinal(self._decode_int() + 719163)
+ elif field_type == TIME:
+ # Time
+ value = self._decode_int()
+ seconds = value // 1000
+ milliseconds = value % 1000
+ minutes = seconds // 60
+ seconds %= 60
+ hours = minutes // 60
+ minutes %= 60
+ return datetime.time(hours, minutes, seconds, milliseconds * 1000)
+
+ cdef unsigned char _decode_byte(self) except? -1:
+ self._input_pos += 1
+ return <unsigned char> self._input_data[self._input_pos - 1]
+
+ cdef libc.stdint.int16_t _decode_smallint(self) except? -1:
+ self._input_pos += 2
+ return (<unsigned char> self._input_data[self._input_pos - 1]
+ | <libc.stdint.uint32_t> <unsigned char> self._input_data[self._input_pos - 2] << 8)
+
+ cdef libc.stdint.int32_t _decode_int(self) except? -1:
+ self._input_pos += 4
+ return (<unsigned char> self._input_data[self._input_pos - 1]
+ | <libc.stdint.uint32_t> <unsigned char> self._input_data[self._input_pos - 2] << 8
+ | <libc.stdint.uint32_t> <unsigned char> self._input_data[self._input_pos - 3] << 16
+ | <libc.stdint.uint32_t> <unsigned char> self._input_data[
+ self._input_pos - 4] << 24)
+
+ cdef libc.stdint.int64_t _decode_bigint(self) except? -1:
+ self._input_pos += 8
+ return (<unsigned char> self._input_data[self._input_pos - 1]
+ | <libc.stdint.uint64_t> <unsigned char> self._input_data[self._input_pos - 2] << 8
+ | <libc.stdint.uint64_t> <unsigned char> self._input_data[self._input_pos - 3] << 16
+ | <libc.stdint.uint64_t> <unsigned char> self._input_data[self._input_pos - 4] << 24
+ | <libc.stdint.uint64_t> <unsigned char> self._input_data[self._input_pos - 5] << 32
+ | <libc.stdint.uint64_t> <unsigned char> self._input_data[self._input_pos - 6] << 40
+ | <libc.stdint.uint64_t> <unsigned char> self._input_data[self._input_pos - 7] << 48
+ | <libc.stdint.uint64_t> <unsigned char> self._input_data[
+ self._input_pos - 8] << 56)
+
+ cdef float _decode_float(self) except? -1:
+ cdef libc.stdint.int32_t as_long = self._decode_int()
+ return (<float*> <char*> &as_long)[0]
+
+ cdef double _decode_double(self) except? -1:
+ cdef libc.stdint.int64_t as_long = self._decode_bigint()
+ return (<double*> <char*> &as_long)[0]
+
+ cdef bytes _decode_bytes(self):
+ cdef libc.stdint.int32_t size = self._decode_int()
+ self._input_pos += size
+ return self._input_data[self._input_pos - size: self._input_pos]
+
+ cdef void _prepare_encode(self, InputStreamAndFunctionWrapper input_stream_and_function_wrapper,
+ OutputStream out_stream):
+ cdef InputStreamWrapper input_stream_wrapper
+ # get the data pointer of output_stream
+ self._output_data = out_stream.data
+ self._output_pos = out_stream.pos
+ self._output_buffer_size = out_stream.buffer_size
+ self._tmp_output_pos = 0
+
+ input_stream_wrapper = input_stream_and_function_wrapper.input_stream_wrapper
+ # get the data pointer of input_stream
+ self._input_data = input_stream_wrapper.input_stream.allc
+ self._input_buffer_size = input_stream_wrapper.input_buffer_size
+
+ # get the infos of input coder which will be used to decode data from input_stream
+ self._input_field_count = input_stream_wrapper.input_field_count
+ self._input_leading_complete_bytes_num = input_stream_wrapper.input_leading_complete_bytes_num
+ self._input_remaining_bits_num = input_stream_wrapper.input_remaining_bits_num
+ self._input_field_type = input_stream_wrapper.input_field_type
+ self._input_coder_type = input_stream_wrapper.input_coder_type
+ self._input_field_coders = input_stream_wrapper.input_field_coders
+ self._null_mask = <bint*> libc.stdlib.malloc(self._input_field_count * sizeof(bint))
+ self._input_pos = 0
+
+ # initial the result row and get the Python user-defined function
+ self.row = [None for _ in range(self._input_field_count)]
+ self.func = input_stream_and_function_wrapper.func
+
+ cdef void _encode_field_simple(self, TypeName field_type, item):
+ cdef libc.stdint.int32_t hour, minute, seconds, microsecond, milliseconds
+ if field_type == TINYINT:
+ # tinyint
+ self._encode_byte(item)
+ elif field_type == SMALLINT:
+ # smallint
+ self._encode_smallint(item)
+ elif field_type == INT:
+ # int
+ self._encode_int(item)
+ elif field_type == BIGINT:
+ # bigint
+ self._encode_bigint(item)
+ elif field_type == BOOLEAN:
+ # boolean
+ self._encode_byte(item)
+ elif field_type == FLOAT:
+ # float
+ self._encode_float(item)
+ elif field_type == DOUBLE:
+ # double
+ self._encode_double(item)
+ elif field_type == BINARY:
+ # bytes
+ self._encode_bytes(item)
+ elif field_type == CHAR:
+ # str
+ self._encode_bytes(item.encode('utf-8'))
+ elif field_type == DATE:
+ # Date
+ # EPOCH_ORDINAL = datetime.datetime(1970, 1, 1).toordinal()
+ # The value of EPOCH_ORDINAL is 719163
+ self._encode_int(item.toordinal() - 719163)
+ elif field_type == TIME:
+ # Time
+ hour = item.hour
+ minute = item.minute
+ seconds = item.second
+ microsecond = item.microsecond
+ milliseconds = hour * 3600000 + minute * 60000 + seconds * 1000 + microsecond // 1000
+ self._encode_int(milliseconds)
+
+ cdef void _copy_to_output_buffer(self):
+ cdef size_t size
+ cdef size_t i
+ cdef bint is_realloc
+ cdef char bits
+ # the length of the variable prefix length will be less than 9 bytes
+ if self._output_buffer_size < self._output_pos + self._tmp_output_pos + 9:
+ self._output_buffer_size += self._tmp_output_buffer_size + 9
+ self._output_data = <char*> libc.stdlib.realloc(self._output_data,
+ self._output_buffer_size)
+ size = self._tmp_output_pos
+ # write variable prefix length
+ while size:
+ bits = size & 0x7F
+ size >>= 7
+ if size:
+ bits |= 0x80
+ self._output_data[self._output_pos] = bits
+ self._output_pos += 1
+ if self._tmp_output_pos < 8:
+ # This is faster than memcpy when the string is short.
+ for i in range(self._tmp_output_pos):
+ self._output_data[self._output_pos + i] = self._tmp_output_data[i]
+ else:
+ libc.string.memcpy(self._output_data + self._output_pos, self._tmp_output_data,
+ self._tmp_output_pos)
+ self._output_pos += self._tmp_output_pos
+ self._tmp_output_pos = 0
+
+ cdef void _maybe_flush(self, OutputStream out_stream):
+ # Currently, it will trigger flushing when the size of buffer reach to 10_000_000
+ if self._output_pos > 10_000_000:
+ self._map_output_data_to_output_stream(out_stream)
+ out_stream.flush()
+ self._output_pos = 0
+
+ cdef void _map_output_data_to_output_stream(self, OutputStream out_stream):
+ out_stream.data = self._output_data
+ out_stream.pos = self._output_pos
+ out_stream.buffer_size = self._output_buffer_size
+
+ cdef void _extend(self, size_t missing):
+ while self._tmp_output_buffer_size < self._tmp_output_pos + missing:
+ self._tmp_output_buffer_size *= 2
+ self._tmp_output_data = <char*> libc.stdlib.realloc(self._tmp_output_data,
+ self._tmp_output_buffer_size)
+
+ cdef void _encode_byte(self, unsigned char val):
+ if self._tmp_output_buffer_size < self._tmp_output_pos + 1:
+ self._extend(1)
+ self._tmp_output_data[self._tmp_output_pos] = val
+ self._tmp_output_pos += 1
+
+ cdef void _encode_smallint(self, libc.stdint.int16_t v):
+ if self._tmp_output_buffer_size < self._tmp_output_pos + 2:
+ self._extend(2)
+ self._tmp_output_data[self._tmp_output_pos] = <unsigned char> (v >> 8)
+ self._tmp_output_data[self._tmp_output_pos + 1] = <unsigned char> v
+ self._tmp_output_pos += 2
+
+ cdef void _encode_int(self, libc.stdint.int32_t v):
+ if self._tmp_output_buffer_size < self._tmp_output_pos + 4:
+ self._extend(4)
+ self._tmp_output_data[self._tmp_output_pos] = <unsigned char> (v >> 24)
+ self._tmp_output_data[self._tmp_output_pos + 1] = <unsigned char> (v >> 16)
+ self._tmp_output_data[self._tmp_output_pos + 2] = <unsigned char> (v >> 8)
+ self._tmp_output_data[self._tmp_output_pos + 3] = <unsigned char> v
+ self._tmp_output_pos += 4
+
+ cdef void _encode_bigint(self, libc.stdint.int64_t v):
+ if self._tmp_output_buffer_size < self._tmp_output_pos + 8:
+ self._extend(8)
+ self._tmp_output_data[self._tmp_output_pos] = <unsigned char> (v >> 56)
+ self._tmp_output_data[self._tmp_output_pos + 1] = <unsigned char> (v >> 48)
+ self._tmp_output_data[self._tmp_output_pos + 2] = <unsigned char> (v >> 40)
+ self._tmp_output_data[self._tmp_output_pos + 3] = <unsigned char> (v >> 32)
+ self._tmp_output_data[self._tmp_output_pos + 4] = <unsigned char> (v >> 24)
+ self._tmp_output_data[self._tmp_output_pos + 5] = <unsigned char> (v >> 16)
+ self._tmp_output_data[self._tmp_output_pos + 6] = <unsigned char> (v >> 8)
+ self._tmp_output_data[self._tmp_output_pos + 7] = <unsigned char> v
+ self._tmp_output_pos += 8
+
+ cdef void _encode_float(self, float v):
+ self._encode_int((<libc.stdint.int32_t*> <char*> &v)[0])
+
+ cdef void _encode_double(self, double v):
+ self._encode_bigint((<libc.stdint.int64_t*> <char*> &v)[0])
+
+ cdef void _encode_bytes(self, char*b):
+ cdef libc.stdint.int32_t length = strlen(b)
+ self._encode_int(length)
+ if self._tmp_output_buffer_size < self._tmp_output_pos + length:
+ self._extend(length)
+ if length < 8:
+ # This is faster than memcpy when the string is short.
+ for i in range(length):
+ self._tmp_output_data[self._tmp_output_pos + i] = b[i]
+ else:
+ libc.string.memcpy(self._tmp_output_data + self._tmp_output_pos, b, length)
+ self._tmp_output_pos += length
+
+ cdef void _write_null_mask(self, value, libc.stdint.int32_t leading_complete_bytes_num,
+ libc.stdint.int32_t remaining_bits_num):
+ cdef libc.stdint.int32_t field_pos, index
+ cdef unsigned char*null_byte_search_table
+ cdef unsigned char b, i
+ field_pos = 0
+ null_byte_search_table = self._null_byte_search_table
+ for _ in range(leading_complete_bytes_num):
+ b = 0x00
+ for i in range(8):
+ if value[field_pos + i] is None:
+ b |= null_byte_search_table[i]
+ field_pos += 8
+ self._encode_byte(b)
+
+ if remaining_bits_num:
+ b = 0x00
+ for i in range(remaining_bits_num):
+ if value[field_pos + i] is None:
+ b |= null_byte_search_table[i]
+ self._encode_byte(b)
+
+ def __dealloc__(self):
+ if self.null_mask:
+ libc.stdlib.free(self._null_mask)
+ if self.null_byte_search_table:
+ libc.stdlib.free(self._null_byte_search_table)
+ if self._tmp_output_data:
+ libc.stdlib.free(self._tmp_output_data)
+ if self._output_field_type:
+ libc.stdlib.free(self._output_field_type)
+ if self._output_coder_type:
+ libc.stdlib.free(self._output_coder_type)
+
+cdef class BaseCoder:
+ cpdef CoderType coder_type(self):
+ return UNDEFINED
+
+ cpdef TypeName type_name(self):
+ return NONE
+
+cdef class TinyIntCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+
+ cpdef TypeName type_name(self):
+ return TINYINT
+
+cdef class SmallIntCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+ cpdef TypeName type_name(self):
+ return SMALLINT
+
+cdef class IntCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+
+ cpdef TypeName type_name(self):
+ return INT
+
+cdef class BigIntCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+ cpdef TypeName type_name(self):
+ return BIGINT
+
+cdef class BooleanCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+ cpdef TypeName type_name(self):
+ return BOOLEAN
+
+cdef class FloatCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+ cpdef TypeName type_name(self):
+ return FLOAT
+
+cdef class DoubleCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+ cpdef TypeName type_name(self):
+ return DOUBLE
+
+cdef class BinaryCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+ cpdef TypeName type_name(self):
+ return BINARY
+
+cdef class CharCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+ cpdef TypeName type_name(self):
+ return CHAR
+
+cdef class DateCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+
+ cpdef TypeName type_name(self):
+ return DATE
+
+cdef class TimeCoderImpl(BaseCoder):
+ cpdef CoderType coder_type(self):
+ return SIMPLE
+
+ cpdef TypeName type_name(self):
+ return TIME
diff --git a/flink-python/pyflink/fn_execution/tests/test_coders_common.py b/flink-python/pyflink/fn_execution/tests/test_coders.py
similarity index 95%
rename from flink-python/pyflink/fn_execution/tests/test_coders_common.py
rename to flink-python/pyflink/fn_execution/tests/test_coders.py
index 69e9961..3134e83 100644
--- a/flink-python/pyflink/fn_execution/tests/test_coders_common.py
+++ b/flink-python/pyflink/fn_execution/tests/test_coders.py
@@ -25,7 +25,16 @@ from pyflink.fn_execution.coders import BigIntCoder, TinyIntCoder, BooleanCoder,
TimeCoder, TimestampCoder, ArrayCoder, MapCoder, DecimalCoder, FlattenRowCoder, RowCoder, \
LocalZonedTimestampCoder
+try:
+ from pyflink.fn_execution import fast_coder_impl # noqa # pylint: disable=unused-import
+ have_cython = True
+except ImportError:
+ have_cython = False
+
+
+@unittest.skipIf(have_cython,
+ "Found cython implementation, we don't need to test non-compiled implementation")
class CodersTest(unittest.TestCase):
def check_coder(self, coder, *values):
diff --git a/flink-python/pyflink/fn_execution/tests/test_fast_coders.py b/flink-python/pyflink/fn_execution/tests/test_fast_coders.py
new file mode 100644
index 0000000..15a8ab0
--- /dev/null
+++ b/flink-python/pyflink/fn_execution/tests/test_fast_coders.py
@@ -0,0 +1,141 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+"""Tests common to all coder implementations."""
+import logging
+import unittest
+
+from pyflink.fn_execution import coder_impl
+
+try:
+ from pyflink.fn_execution import fast_coder_impl
+
+ have_cython = True
+except ImportError:
+ have_cython = False
+
+
+@unittest.skipUnless(have_cython, "Uncompiled Cython Coder")
+class CodersTest(unittest.TestCase):
+
+ def check_cython_coder(self, python_field_coders, cython_field_coders, data):
+ from apache_beam.coders.coder_impl import create_InputStream, create_OutputStream
+ from pyflink.fn_execution.fast_coder_impl import InputStreamAndFunctionWrapper
+ py_flatten_row_coder = coder_impl.FlattenRowCoderImpl(python_field_coders)
+ internal = py_flatten_row_coder.encode(data)
+ input_stream = create_InputStream(internal)
+ output_stream = create_OutputStream()
+ cy_flatten_row_coder = fast_coder_impl.FlattenRowCoderImpl(cython_field_coders)
+ value = cy_flatten_row_coder.decode_from_stream(input_stream, False)
+ wrapper_func_input_element = InputStreamAndFunctionWrapper(
+ lambda v: [v[i] for i in range(len(v))], value)
+ cy_flatten_row_coder.encode_to_stream(wrapper_func_input_element, output_stream, False)
+ generator_result = py_flatten_row_coder.decode_from_stream(create_InputStream(
+ output_stream.get()), False)
+ result = []
+ for item in generator_result:
+ result.append(item)
+ try:
+ self.assertEqual(result, data)
+ except AssertionError:
+ self.assertEqual(len(result), len(data))
+ self.assertEqual(len(result[0]), len(data[0]))
+ for i in range(len(data[0])):
+ if isinstance(data[0][i], float):
+ from pyflink.table.tests.test_udf import float_equal
+ assert float_equal(data[0][i], result[0][i], 1e-6)
+ else:
+ self.assertEqual(data[0][i], result[0][i])
+
+ # decide whether two floats are equal
+ @staticmethod
+ def float_equal(a, b, rel_tol=1e-09, abs_tol=0.0):
+ return abs(a - b) <= max(rel_tol * max(abs(a), abs(b)), abs_tol)
+
+ def test_cython_bigint_coder(self):
+ data = [1, 100, -100, -1000]
+ python_field_coders = [coder_impl.BigIntCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.BigIntCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_tinyint_coder(self):
+ data = [1, 10, 127, -128]
+ python_field_coders = [coder_impl.TinyIntCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.TinyIntCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_boolean_coder(self):
+ data = [True, False]
+ python_field_coders = [coder_impl.BooleanCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.BooleanCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_smallint_coder(self):
+ data = [32767, -32768, 0]
+ python_field_coders = [coder_impl.SmallIntCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.SmallIntCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_int_coder(self):
+ data = [-2147483648, 2147483647]
+ python_field_coders = [coder_impl.IntCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.IntCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_float_coder(self):
+ data = [1.02, 1.32]
+ python_field_coders = [coder_impl.FloatCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.FloatCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_double_coder(self):
+ data = [-12.02, 1.98932]
+ python_field_coders = [coder_impl.DoubleCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.DoubleCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_binary_coder(self):
+ data = [b'pyflink']
+ python_field_coders = [coder_impl.BinaryCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.BinaryCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_char_coder(self):
+ data = ['flink', '🐿']
+ python_field_coders = [coder_impl.CharCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.CharCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_date_coder(self):
+ import datetime
+ data = [datetime.date(2019, 9, 10)]
+ python_field_coders = [coder_impl.DateCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.DateCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+ def test_cython_time_coder(self):
+ import datetime
+ data = [datetime.time(hour=11, minute=11, second=11, microsecond=123000)]
+ python_field_coders = [coder_impl.TimeCoderImpl() for _ in range(len(data))]
+ cython_field_coders = [fast_coder_impl.TimeCoderImpl() for _ in range(len(data))]
+ self.check_cython_coder(python_field_coders, cython_field_coders, [data])
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/flink-python/setup.py b/flink-python/setup.py
index c74abeb..4fa842e 100644
--- a/flink-python/setup.py
+++ b/flink-python/setup.py
@@ -20,9 +20,10 @@ from __future__ import print_function
import io
import os
import sys
+from distutils.command.build_ext import build_ext
from shutil import copytree, copy, rmtree
-from setuptools import setup
+from setuptools import setup, Extension
if sys.version_info < (3, 5):
print("Python versions prior to 3.5 are not supported for PyFlink.",
@@ -39,6 +40,26 @@ def remove_if_exists(file_path):
rmtree(file_path)
+try:
+ from Cython.Build import cythonize
+ extensions = cythonize([
+ Extension(
+ name="pyflink.fn_execution.fast_coder_impl",
+ sources=["pyflink/fn_execution/fast_coder_impl.pyx"],
+ include_dirs=["pyflink/fn_execution/"])
+ ])
+except ImportError:
+ if os.path.exists("pyflink/fn_execution/fast_coder_impl.c"):
+ extensions = ([
+ Extension(
+ name="pyflink.fn_execution.fast_coder_impl",
+ sources=["pyflink/fn_execution/fast_coder_impl.c"],
+ include_dirs=["pyflink/fn_execution/"])
+ ])
+ else:
+ extensions = ([])
+
+
this_directory = os.path.abspath(os.path.dirname(__file__))
version_file = os.path.join(this_directory, 'pyflink/version.py')
@@ -232,16 +253,19 @@ run sdist.
install_requires=['py4j==0.10.8.1', 'python-dateutil==2.8.0', 'apache-beam==2.19.0',
'cloudpickle==1.2.2', 'avro-python3>=1.8.1,<=1.9.1', 'jsonpickle==1.2',
'pandas>=0.23.4,<=0.25.3', 'pyarrow>=0.15.1,<0.16.0', 'pytz>=2018.3'],
+ cmdclass={'build_ext': build_ext},
tests_require=['pytest==4.4.1'],
description='Apache Flink Python API',
long_description=long_description,
long_description_content_type='text/markdown',
+ zip_safe=False,
classifiers=[
'Development Status :: 5 - Production/Stable',
'License :: OSI Approved :: Apache Software License',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
- 'Programming Language :: Python :: 3.7']
+ 'Programming Language :: Python :: 3.7'],
+ ext_modules=extensions
)
finally:
if in_flink_source:
diff --git a/flink-python/tox.ini b/flink-python/tox.ini
index b944857..47ea7d8 100644
--- a/flink-python/tox.ini
+++ b/flink-python/tox.ini
@@ -21,17 +21,23 @@
# in multiple virtualenvs. This configuration file will run the
# test suite on all supported python versions.
# new environments will be excluded by default unless explicitly added to envlist.
-envlist = py35, py36, py37
+envlist = {py35, py36, py37}-cython
[testenv]
whitelist_externals=
/bin/bash
deps =
pytest
+ apache-beam==2.19.0
+ cython==0.29.16
grpcio>=1.17.0,<=1.26.0
grpcio-tools>=1.3.5,<=1.14.2
commands =
python --version
+ # python test
+ pytest --durations=0
+ python setup.py build_ext --inplace
+ # cython test
pytest --durations=0
bash ./dev/run_pip_test.sh
# Replace the default installation command with a custom retry installation script, because on high-speed