Source code for pyflink.table.table_schema

################################################################################
#  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.
################################################################################
from typing import List, Optional, Union

from pyflink.java_gateway import get_gateway
from pyflink.table.types import _to_java_type, _from_java_type, DataType, RowType
from pyflink.util.java_utils import to_jarray

__all__ = ['TableSchema']


[docs]class TableSchema(object): """ A table schema that represents a table's structure with field names and data types. """ def __init__(self, field_names: List[str] = None, data_types: List[DataType] = None, j_table_schema=None): if j_table_schema is None: gateway = get_gateway() j_field_names = to_jarray(gateway.jvm.String, field_names) j_data_types = to_jarray(gateway.jvm.TypeInformation, [_to_java_type(item) for item in data_types]) self._j_table_schema = gateway.jvm.TableSchema(j_field_names, j_data_types) else: self._j_table_schema = j_table_schema
[docs] def copy(self) -> 'TableSchema': """ Returns a deep copy of the table schema. :return: A deep copy of the table schema. """ return TableSchema(j_table_schema=self._j_table_schema.copy())
[docs] def get_field_data_types(self) -> List[DataType]: """ Returns all field data types as a list. :return: A list of all field data types. """ return [_from_java_type(item) for item in self._j_table_schema.getFieldDataTypes()]
[docs] def get_field_data_type(self, field: Union[int, str]) -> Optional[DataType]: """ Returns the specified data type for the given field index or field name. :param field: The index of the field or the name of the field. :return: The data type of the specified field. """ if not isinstance(field, (int, str)): raise TypeError("Expected field index or field name, got %s" % type(field)) optional_result = self._j_table_schema.getFieldDataType(field) if optional_result.isPresent(): return _from_java_type(optional_result.get()) else: return None
[docs] def get_field_count(self) -> int: """ Returns the number of fields. :return: The number of fields. """ return self._j_table_schema.getFieldCount()
[docs] def get_field_names(self) -> List[str]: """ Returns all field names as a list. :return: The list of all field names. """ return list(self._j_table_schema.getFieldNames())
[docs] def get_field_name(self, field_index: int) -> Optional[str]: """ Returns the specified name for the given field index. :param field_index: The index of the field. :return: The field name. """ optional_result = self._j_table_schema.getFieldName(field_index) if optional_result.isPresent(): return optional_result.get() else: return None
[docs] def to_row_data_type(self) -> RowType: """ Converts a table schema into a (nested) data type describing a :func:`pyflink.table.types.DataTypes.ROW`. :return: The row data type. """ return _from_java_type(self._j_table_schema.toRowDataType())
def __repr__(self): return self._j_table_schema.toString() def __eq__(self, other): return isinstance(other, self.__class__) and self._j_table_schema == other._j_table_schema def __hash__(self): return self._j_table_schema.hashCode() def __ne__(self, other): return not self.__eq__(other)
[docs] @classmethod def builder(cls): return TableSchema.Builder()
[docs] class Builder(object): """ Builder for creating a :class:`TableSchema`. """ def __init__(self): self._field_names = [] self._field_data_types = []
[docs] def field(self, name: str, data_type: DataType) -> 'TableSchema.Builder': """ Add a field with name and data type. The call order of this method determines the order of fields in the schema. :param name: The field name. :param data_type: The field data type. :return: This object. """ assert name is not None assert data_type is not None self._field_names.append(name) self._field_data_types.append(data_type) return self
[docs] def build(self) -> 'TableSchema': """ Returns a :class:`TableSchema` instance. :return: The :class:`TableSchema` instance. """ return TableSchema(self._field_names, self._field_data_types)