| ################################################################################ |
| # 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. |
| ################################################################################ |
| import calendar |
| import datetime |
| import time |
| from abc import ABC |
| from typing import List, Union |
| |
| from py4j.java_gateway import JavaClass, JavaObject |
| |
| from pyflink.java_gateway import get_gateway |
| |
| |
| class TypeInformation(ABC): |
| """ |
| 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 exist. |
| |
| The type information also bridges between the programming languages object model and a logical |
| flat schema. It maps fields from the types to columns (fields) in a flat schema. Not all fields |
| from a type are mapped to a separate fields in the flat schema and often, entire types are |
| mapped to one field.. It is important to notice that the schema must hold for all instances of a |
| type. For that reason, elements in lists and arrays are not assigned to individual fields, but |
| the lists and arrays are considered to be one field in total, to account for different lengths |
| in the arrays. |
| a) Basic types are indivisible and are considered as a single field. |
| b) Arrays and collections are one field. |
| c) Tuples represents as many fields as the class has fields. |
| To represent this properly, each type has an arity (the number of fields it contains directly), |
| and a total number of fields (number of fields in the entire schema of this type, including |
| nested types). |
| """ |
| |
| |
| class WrapperTypeInfo(TypeInformation): |
| """ |
| A wrapper class for java TypeInformation Objects. |
| """ |
| |
| def __init__(self, j_typeinfo): |
| self._j_typeinfo = j_typeinfo |
| |
| def get_java_type_info(self) -> JavaObject: |
| return self._j_typeinfo |
| |
| def __eq__(self, o) -> bool: |
| if type(o) is type(self): |
| return self._j_typeinfo.equals(o._j_typeinfo) |
| else: |
| return False |
| |
| def __hash__(self) -> int: |
| return hash(self._j_typeinfo) |
| |
| def __str__(self): |
| return self._j_typeinfo.toString() |
| |
| def need_conversion(self): |
| """ |
| Does this type need to conversion between Python object and internal Wrapper object. |
| """ |
| return False |
| |
| def to_internal_type(self, obj): |
| """ |
| Converts a Python object into an internal object. |
| """ |
| return obj |
| |
| |
| class BasicTypeInfo(TypeInformation, ABC): |
| """ |
| Type information for primitive types (int, long, double, byte, ...), String, BigInteger, |
| and BigDecimal. |
| """ |
| |
| @staticmethod |
| def STRING_TYPE_INFO(): |
| return WrapperTypeInfo(get_gateway().jvm |
| .org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO) |
| |
| @staticmethod |
| def BOOLEAN_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.BOOLEAN_TYPE_INFO) |
| |
| @staticmethod |
| def BYTE_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.BYTE_TYPE_INFO) |
| |
| @staticmethod |
| def SHORT_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.SHORT_TYPE_INFO) |
| |
| @staticmethod |
| def INT_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO) |
| |
| @staticmethod |
| def LONG_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.LONG_TYPE_INFO) |
| |
| @staticmethod |
| def FLOAT_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.FLOAT_TYPE_INFO) |
| |
| @staticmethod |
| def DOUBLE_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.DOUBLE_TYPE_INFO) |
| |
| @staticmethod |
| def CHAR_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.CHAR_TYPE_INFO) |
| |
| @staticmethod |
| def BIG_INT_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.BIG_INT_TYPE_INFO) |
| |
| @staticmethod |
| def BIG_DEC_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo.BIG_DEC_TYPE_INFO) |
| |
| |
| class SqlTimeTypeInfo(TypeInformation, ABC): |
| """ |
| SqlTimeTypeInfo enables users to get Sql Time TypeInfo. |
| """ |
| |
| @staticmethod |
| def DATE(): |
| return DateTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo.DATE) |
| |
| @staticmethod |
| def TIME(): |
| return TimeTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo.TIME) |
| |
| @staticmethod |
| def TIMESTAMP(): |
| return TimestampTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo.TIMESTAMP) |
| |
| |
| class PrimitiveArrayTypeInfo(WrapperTypeInfo, ABC): |
| """ |
| A TypeInformation for arrays of primitive types (int, long, double, ...). |
| Supports the creation of dedicated efficient serializers for these types. |
| """ |
| |
| @staticmethod |
| def BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def BYTE_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def SHORT_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def INT_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def LONG_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def FLOAT_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def CHAR_PRIMITIVE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .PrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO) |
| |
| |
| class BasicArrayTypeInfo(WrapperTypeInfo, ABC): |
| """ |
| A TypeInformation for arrays of boxed primitive types (Integer, Long, Double, ...). |
| Supports the creation of dedicated efficient serializers for these types. |
| """ |
| @staticmethod |
| def BOOLEAN_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.BOOLEAN_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def BYTE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def SHORT_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.SHORT_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def INT_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def LONG_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.LONG_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def FLOAT_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.FLOAT_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def DOUBLE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.DOUBLE_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def CHAR_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.CHAR_ARRAY_TYPE_INFO) |
| |
| @staticmethod |
| def STRING_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo |
| .BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO) |
| |
| |
| class PickledBytesTypeInfo(WrapperTypeInfo, ABC): |
| """ |
| A PickledBytesTypeInfo indicates the data is a primitive byte array generated by pickle |
| serializer. |
| """ |
| |
| @staticmethod |
| def PICKLED_BYTE_ARRAY_TYPE_INFO(): |
| return WrapperTypeInfo(get_gateway().jvm.org.apache.flink.streaming.api.typeinfo.python |
| .PickledByteArrayTypeInfo.PICKLED_BYTE_ARRAY_TYPE_INFO) |
| |
| |
| class RowTypeInfo(WrapperTypeInfo): |
| """ |
| TypeInformation for Row. |
| """ |
| |
| def __init__(self, types: List[TypeInformation], field_names: List[str] = None): |
| self.types = types |
| self.field_names = field_names |
| self.j_types_array = get_gateway().new_array( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.TypeInformation, len(types)) |
| for i in range(len(types)): |
| wrapper_typeinfo = types[i] |
| if isinstance(wrapper_typeinfo, WrapperTypeInfo): |
| self.j_types_array[i] = wrapper_typeinfo.get_java_type_info() |
| |
| if field_names is None: |
| self._j_typeinfo = get_gateway().jvm.org.apache.flink.api.java.typeutils.RowTypeInfo( |
| self.j_types_array) |
| else: |
| j_names_array = get_gateway().new_array(get_gateway().jvm.java.lang.String, |
| len(field_names)) |
| for i in range(len(field_names)): |
| j_names_array[i] = field_names[i] |
| self._j_typeinfo = get_gateway().jvm.org.apache.flink.api.java.typeutils.RowTypeInfo( |
| self.j_types_array, j_names_array) |
| self._need_conversion = [f.need_conversion() if isinstance(f, WrapperTypeInfo) else None |
| for f in types] |
| self._need_serialize_any_field = any(self._need_conversion) |
| super(RowTypeInfo, self).__init__(self._j_typeinfo) |
| |
| def get_field_names(self) -> List[str]: |
| j_field_names = self._j_typeinfo.getFieldNames() |
| field_names = [name for name in j_field_names] |
| return field_names |
| |
| def get_field_index(self, field_name: str) -> int: |
| return self._j_typeinfo.getFieldIndex(field_name) |
| |
| def get_field_types(self) -> List[TypeInformation]: |
| return self.types |
| |
| def __eq__(self, other) -> bool: |
| return self._j_typeinfo.equals(other._j_typeinfo) |
| |
| def __hash__(self) -> int: |
| return self._j_typeinfo.hashCode() |
| |
| def __str__(self) -> str: |
| |
| return "RowTypeInfo(%s)" % ', '.join([field_name + ': ' + field_type.__str__() |
| for field_name, field_type in |
| zip(self.get_field_names(), |
| self.get_field_types())]) |
| |
| def need_conversion(self): |
| return True |
| |
| def to_internal_type(self, obj): |
| if obj is None: |
| return |
| |
| if self._need_serialize_any_field: |
| # Only calling to_internal_type function for fields that need conversion |
| if isinstance(obj, dict): |
| return tuple(f.to_internal_type(obj.get(n)) if c else obj.get(n) |
| for n, f, c in zip(self._j_typeinfo.getFieldNames(), self.types, |
| self._need_conversion)) |
| elif isinstance(obj, (tuple, list)): |
| return tuple(f.to_internal_type(v) if c else v |
| for f, v, c in zip(self.types, obj, self._need_conversion)) |
| elif hasattr(obj, "__dict__"): |
| d = obj.__dict__ |
| return tuple(f.to_internal_type(d.get(n)) if c else d.get(n) |
| for n, f, c in zip(self._j_typeinfo.getFieldNames(), self.types, |
| self._need_conversion)) |
| else: |
| raise ValueError("Unexpected tuple %r with RowTypeInfo" % obj) |
| else: |
| if isinstance(obj, dict): |
| return tuple(obj.get(n) for n in self._j_typeinfo.getFieldNames()) |
| elif isinstance(obj, (list, tuple)): |
| return tuple(obj) |
| elif hasattr(obj, "__dict__"): |
| d = obj.__dict__ |
| return tuple(d.get(n) for n in self._j_typeinfo.getFieldNames()) |
| else: |
| raise ValueError("Unexpected tuple %r with RowTypeInfo" % obj) |
| |
| |
| class TupleTypeInfo(WrapperTypeInfo): |
| """ |
| TypeInformation for Tuple. |
| """ |
| |
| def __init__(self, types: List[TypeInformation]): |
| self.types = types |
| j_types_array = get_gateway().new_array( |
| get_gateway().jvm.org.apache.flink.api.common.typeinfo.TypeInformation, len(types)) |
| |
| for i in range(len(types)): |
| field_type = types[i] |
| if isinstance(field_type, WrapperTypeInfo): |
| j_types_array[i] = field_type.get_java_type_info() |
| |
| j_typeinfo = get_gateway().jvm \ |
| .org.apache.flink.api.java.typeutils.TupleTypeInfo(j_types_array) |
| super(TupleTypeInfo, self).__init__(j_typeinfo=j_typeinfo) |
| |
| def get_field_types(self) -> List[TypeInformation]: |
| return self.types |
| |
| def __eq__(self, other) -> bool: |
| return self._j_typeinfo.equals(other._j_typeinfo) |
| |
| def __hash__(self) -> int: |
| return self._j_typeinfo.hashCode() |
| |
| def __str__(self) -> str: |
| return "TupleTypeInfo(%s)" % ', '.join([field_type.__str__() for field_type in self.types]) |
| |
| |
| class DateTypeInfo(WrapperTypeInfo): |
| """ |
| TypeInformation for Date. |
| """ |
| |
| def __init__(self, j_typeinfo): |
| super(DateTypeInfo, self).__init__(j_typeinfo) |
| |
| EPOCH_ORDINAL = datetime.datetime(1970, 1, 1).toordinal() |
| |
| def need_conversion(self): |
| return True |
| |
| def to_internal_type(self, d): |
| if d is not None: |
| return d.toordinal() - self.EPOCH_ORDINAL |
| |
| |
| class TimeTypeInfo(WrapperTypeInfo): |
| """ |
| TypeInformation for Time. |
| """ |
| |
| EPOCH_ORDINAL = calendar.timegm(time.localtime(0)) * 10 ** 6 |
| |
| def __init__(self, j_typeinfo): |
| super(TimeTypeInfo, self).__init__(j_typeinfo) |
| |
| def need_conversion(self): |
| return True |
| |
| def to_internal_type(self, t): |
| if t is not None: |
| if t.tzinfo is not None: |
| offset = t.utcoffset() |
| offset = offset if offset else datetime.timedelta() |
| offset_microseconds =\ |
| (offset.days * 86400 + offset.seconds) * 10 ** 6 + offset.microseconds |
| else: |
| offset_microseconds = self.EPOCH_ORDINAL |
| minutes = t.hour * 60 + t.minute |
| seconds = minutes * 60 + t.second |
| return seconds * 10 ** 6 + t.microsecond - offset_microseconds |
| |
| |
| class TimestampTypeInfo(WrapperTypeInfo): |
| """ |
| TypeInformation for Timestamp. |
| """ |
| |
| def __init__(self, j_typeinfo): |
| super(TimestampTypeInfo, self).__init__(j_typeinfo) |
| |
| def need_conversion(self): |
| return True |
| |
| def to_internal_type(self, dt): |
| if dt is not None: |
| seconds = (calendar.timegm(dt.utctimetuple()) if dt.tzinfo |
| else time.mktime(dt.timetuple())) |
| return int(seconds) * 10 ** 6 + dt.microsecond |
| |
| |
| class Types(object): |
| """ |
| This class gives access to the type information of the most common types for which Flink has |
| built-in serializers and comparators. |
| """ |
| |
| STRING = BasicTypeInfo.STRING_TYPE_INFO |
| BYTE = BasicTypeInfo.BYTE_TYPE_INFO |
| BOOLEAN = BasicTypeInfo.BOOLEAN_TYPE_INFO |
| SHORT = BasicTypeInfo.SHORT_TYPE_INFO |
| INT = BasicTypeInfo.INT_TYPE_INFO |
| LONG = BasicTypeInfo.LONG_TYPE_INFO |
| FLOAT = BasicTypeInfo.FLOAT_TYPE_INFO |
| DOUBLE = BasicTypeInfo.DOUBLE_TYPE_INFO |
| CHAR = BasicTypeInfo.CHAR_TYPE_INFO |
| BIG_INT = BasicTypeInfo.BIG_INT_TYPE_INFO |
| BIG_DEC = BasicTypeInfo.BIG_DEC_TYPE_INFO |
| |
| SQL_DATE = SqlTimeTypeInfo.DATE |
| SQL_TIME = SqlTimeTypeInfo.TIME |
| SQL_TIMESTAMP = SqlTimeTypeInfo.TIMESTAMP |
| |
| PICKLED_BYTE_ARRAY = PickledBytesTypeInfo.PICKLED_BYTE_ARRAY_TYPE_INFO |
| |
| @staticmethod |
| def ROW(types: List[TypeInformation]): |
| """ |
| Returns type information for Row with fields of the given types. A row itself must not be |
| null. |
| |
| :param types: the types of the row fields, e.g., Types.String(), Types.INT() |
| """ |
| return RowTypeInfo(types) |
| |
| @staticmethod |
| def ROW_NAMED(names: List[str], types: List[TypeInformation]): |
| """ |
| Returns type information for Row with fields of the given types and with given names. A row |
| must not be null. |
| |
| :param names: array of field names. |
| :param types: array of field types. |
| """ |
| return RowTypeInfo(types, names) |
| |
| @staticmethod |
| def TUPLE(types: List[TypeInformation]): |
| """ |
| Returns type information for Tuple with fields of the given types. A Tuple itself must not |
| be null. |
| |
| :param types: array of field types. |
| """ |
| return TupleTypeInfo(types) |
| |
| @staticmethod |
| def PRIMITIVE_ARRAY(element_type: TypeInformation): |
| """ |
| Returns type information for arrays of primitive type (such as byte[]). The array must not |
| be null. |
| |
| :param element_type element type of the array (e.g. Types.BOOLEAN(), Types.INT(), |
| Types.DOUBLE()) |
| """ |
| if element_type == Types.BOOLEAN(): |
| return PrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO() |
| elif element_type == Types.BYTE(): |
| return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO() |
| elif element_type == Types.SHORT(): |
| return PrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO() |
| elif element_type == Types.INT(): |
| return PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO() |
| elif element_type == Types.LONG(): |
| return PrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO() |
| elif element_type == Types.FLOAT(): |
| return PrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO() |
| elif element_type == Types.DOUBLE(): |
| return PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO() |
| elif element_type == Types.CHAR(): |
| return PrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO() |
| else: |
| raise TypeError("Invalid element type for a primitive array.") |
| |
| @staticmethod |
| def BASIC_ARRAY(element_type: TypeInformation) -> TypeInformation: |
| """ |
| Returns type information for arrays of boxed primitive type (such as Integer[]). |
| |
| :param element_type element type of the array (e.g. Types.BOOLEAN(), Types.INT(), |
| Types.DOUBLE()) |
| """ |
| if element_type == Types.BOOLEAN(): |
| return BasicArrayTypeInfo.BOOLEAN_ARRAY_TYPE_INFO() |
| elif element_type == Types.BYTE(): |
| return BasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO() |
| elif element_type == Types.SHORT(): |
| return BasicArrayTypeInfo.SHORT_ARRAY_TYPE_INFO() |
| elif element_type == Types.INT(): |
| return BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO() |
| elif element_type == Types.LONG(): |
| return BasicArrayTypeInfo.LONG_ARRAY_TYPE_INFO() |
| elif element_type == Types.FLOAT(): |
| return BasicArrayTypeInfo.FLOAT_ARRAY_TYPE_INFO() |
| elif element_type == Types.DOUBLE(): |
| return BasicArrayTypeInfo.DOUBLE_ARRAY_TYPE_INFO() |
| elif element_type == Types.CHAR(): |
| return BasicArrayTypeInfo.CHAR_ARRAY_TYPE_INFO() |
| elif element_type == Types.STRING(): |
| return BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO() |
| else: |
| raise TypeError("Invalid element type for a boxed primitive array: %s" % |
| str(element_type)) |
| |
| |
| def _from_java_type(j_type_info: JavaObject) -> TypeInformation: |
| gateway = get_gateway() |
| JBasicTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo |
| |
| if _is_instance_of(j_type_info, JBasicTypeInfo.STRING_TYPE_INFO): |
| return Types.STRING() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.BOOLEAN_TYPE_INFO): |
| return Types.BOOLEAN() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.BYTE_TYPE_INFO): |
| return Types.BYTE() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.SHORT_TYPE_INFO): |
| return Types.SHORT() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.INT_TYPE_INFO): |
| return Types.INT() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.LONG_TYPE_INFO): |
| return Types.LONG() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.FLOAT_TYPE_INFO): |
| return Types.FLOAT() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.DOUBLE_TYPE_INFO): |
| return Types.DOUBLE() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.CHAR_TYPE_INFO): |
| return Types.CHAR() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_INT_TYPE_INFO): |
| return Types.BIG_INT() |
| elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_DEC_TYPE_INFO): |
| return Types.BIG_DEC() |
| |
| JSqlTimeTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo |
| if _is_instance_of(j_type_info, JSqlTimeTypeInfo.DATE): |
| return Types.SQL_DATE() |
| elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIME): |
| return Types.SQL_TIME() |
| elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIMESTAMP): |
| return Types.SQL_TIMESTAMP() |
| |
| JPrimitiveArrayTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo \ |
| .PrimitiveArrayTypeInfo |
| |
| if _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.BOOLEAN()) |
| elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.BYTE()) |
| elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.SHORT()) |
| elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.INT()) |
| elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.LONG()) |
| elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.FLOAT()) |
| elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.DOUBLE()) |
| elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO): |
| return Types.PRIMITIVE_ARRAY(Types.CHAR()) |
| |
| JBasicArrayTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo \ |
| .BasicArrayTypeInfo |
| |
| if _is_instance_of(j_type_info, JBasicArrayTypeInfo.BOOLEAN_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.BOOLEAN()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.BYTE()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.SHORT_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.SHORT()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.INT_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.INT()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.LONG_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.LONG()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.FLOAT_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.FLOAT()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.DOUBLE_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.DOUBLE()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.CHAR_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.CHAR()) |
| elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO): |
| return Types.BASIC_ARRAY(Types.STRING()) |
| |
| JPickledBytesTypeInfo = gateway.jvm \ |
| .org.apache.flink.streaming.api.typeinfo.python.PickledByteArrayTypeInfo\ |
| .PICKLED_BYTE_ARRAY_TYPE_INFO |
| if _is_instance_of(j_type_info, JPickledBytesTypeInfo): |
| return Types.PICKLED_BYTE_ARRAY() |
| |
| JRowTypeInfo = gateway.jvm.org.apache.flink.api.java.typeutils.RowTypeInfo |
| if _is_instance_of(j_type_info, JRowTypeInfo): |
| j_row_field_names = j_type_info.getFieldNames() |
| j_row_field_types = j_type_info.getFieldTypes() |
| row_field_types = [_from_java_type(j_row_field_type) for j_row_field_type in |
| j_row_field_types] |
| return Types.ROW_NAMED(j_row_field_names, row_field_types) |
| |
| JTupleTypeInfo = gateway.jvm.org.apache.flink.api.java.typeutils.TupleTypeInfo |
| if _is_instance_of(j_type_info, JTupleTypeInfo): |
| j_field_types = [] |
| for i in range(j_type_info.getArity()): |
| j_field_types.append(j_type_info.getTypeAt(i)) |
| field_types = [_from_java_type(j_field_type) for j_field_type in j_field_types] |
| return TupleTypeInfo(field_types) |
| |
| 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: |
| if isinstance(java_type, JavaObject): |
| return java_object.equals(java_type) |
| elif isinstance(java_type, JavaClass): |
| return java_object.getClass().isAssignableFrom(java_type._java_lang_class) |
| return False |
