[ https://issues.apache.org/jira/browse/FLINK-7452?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16291071#comment-16291071 ]
ASF GitHub Bot commented on FLINK-7452: --------------------------------------- Github user fhueske commented on a diff in the pull request: https://github.com/apache/flink/pull/4612#discussion_r156958975 --- Diff: flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/Types.scala --- @@ -0,0 +1,371 @@ +/* + * 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. + */ + +package org.apache.flink.api.scala.typeutils + +import org.apache.flink.annotation.PublicEvolving +import org.apache.flink.api.common.functions.InvalidTypesException +import org.apache.flink.api.common.typeinfo.{TypeInformation, Types => JTypes} +import org.apache.flink.types.Row + +import _root_.scala.collection.JavaConverters._ +import _root_.scala.util.{Either, Try} + +/** + * This class gives access to the type information of the most common Scala types for which Flink + * has built-in serializers and comparators. + * + * This class contains types of [[org.apache.flink.api.common.typeinfo.Types]] and adds + * types for Scala specific classes (such as [[Unit]] or case classes). + * + * In many cases, Flink tries to analyze generic signatures of functions to determine return + * types automatically. This class is intended for cases where type information has to be + * supplied manually or would result in an inefficient type. + * + * Scala macros allow to determine type information of classes and type parameters. You can + * use [[Types.of]] to let type information be determined automatically. + */ +@PublicEvolving +object Types { + + /** + * Generates type information based on the given class and/or its type parameters. + * + * The definition is similar to a [[org.apache.flink.api.common.typeinfo.TypeHint]] but does + * not require to implement anonymous classes. + * + * If the class could not be analyzed by the Scala type analyzer, the Java analyzer + * will be used. + * + * Example use: + * + * `Types.of[(Int, String, String)]` for Scala tuples + * `Types.of[Unit]` for Scala specific types + * + * @tparam T class to be analyzed + */ + def of[T: TypeInformation]: TypeInformation[T] = { + val typeInfo: TypeInformation[T] = implicitly[TypeInformation[T]] + typeInfo + } + + /** + * Returns type information for Scala [[Nothing]]. Does not support a null value. + */ + val NOTHING: TypeInformation[Nothing] = new ScalaNothingTypeInfo + + /** + * Returns type information for Scala [[Unit]]. Does not support a null value. + */ + val UNIT: TypeInformation[Unit] = new UnitTypeInfo + + /** + * Returns type information for [[String]] and [[java.lang.String]]. Supports a null value. + */ + val STRING: TypeInformation[String] = JTypes.STRING + + /** + * Returns type information for primitive [[Byte]] and [[java.lang.Byte]]. Does not + * support a null value. + */ + val BYTE: TypeInformation[java.lang.Byte] = JTypes.BYTE + + /** + * Returns type information for primitive [[Boolean]] and [[java.lang.Boolean]]. Does not + * support a null value. + */ + val BOOLEAN: TypeInformation[java.lang.Boolean] = JTypes.BOOLEAN + + /** + * Returns type information for primitive [[Short]] and [[java.lang.Short]]. Does not + * support a null value. + */ + val SHORT: TypeInformation[java.lang.Short] = JTypes.SHORT + + /** + * Returns type information for primitive [[Int]] and [[java.lang.Integer]]. Does not + * support a null value. + */ + val INT: TypeInformation[java.lang.Integer] = JTypes.INT + + /** + * Returns type information for primitive [[Long]] and [[java.lang.Long]]. Does not + * support a null value. + */ + val LONG: TypeInformation[java.lang.Long] = JTypes.LONG + + /** + * Returns type information for primitive [[Float]] and [[java.lang.Float]]. Does not + * support a null value. + */ + val FLOAT: TypeInformation[java.lang.Float] = JTypes.FLOAT + + /** + * Returns type information for primitive [[Double]] and [[java.lang.Double]]. Does not + * support a null value. + */ + val DOUBLE: TypeInformation[java.lang.Double] = JTypes.DOUBLE + + /** + * Returns type information for primitive [[Char]] and [[java.lang.Character]]. Does not + * support a null value. + */ + val CHAR: TypeInformation[java.lang.Character] = JTypes.CHAR + + /** + * Returns type information for Java [[java.math.BigDecimal]]. Supports a null value. + * + * Note that Scala [[BigDecimal]] is not supported yet. + */ + val JAVA_BIG_DEC: TypeInformation[java.math.BigDecimal] = JTypes.BIG_DEC + + /** + * Returns type information for Java [[java.math.BigInteger]]. Supports a null value. + * + * Note that Scala [[BigInt]] is not supported yet. + */ + val JAVA_BIG_INT: TypeInformation[java.math.BigInteger] = JTypes.BIG_INT + + /** + * Returns type information for [[java.sql.Date]]. Supports a null value. + */ + val SQL_DATE: TypeInformation[java.sql.Date] = JTypes.SQL_DATE + + /** + * Returns type information for [[java.sql.Time]]. Supports a null value. + */ + val SQL_TIME: TypeInformation[java.sql.Time] = JTypes.SQL_TIME + + /** + * Returns type information for [[java.sql.Timestamp]]. Supports a null value. + */ + val SQL_TIMESTAMP: TypeInformation[java.sql.Timestamp] = JTypes.SQL_TIMESTAMP + + /** + * Returns type information for [[org.apache.flink.types.Row]] with fields of the given types. + * A row itself must not be null. + * + * A row is a fixed-length, null-aware composite type for storing multiple values in a + * deterministic field order. Every field can be null independent of the field's type. + * The type of row fields cannot be automatically inferred; therefore, it is required to pass + * type information whenever a row is used. + * + * <p>The schema of rows can have up to <code>Integer.MAX_VALUE</code> fields, however, all row instances + * must have the same length otherwise serialization fails or information is lost. + * + * This method generates type information with fields of the given types; the fields have + * the default names (f0, f1, f2 ..). + * + * @param types The types of the row fields, e.g., Types.STRING, Types.INT + */ + def ROW(types: TypeInformation[_]*): TypeInformation[Row] = JTypes.ROW(types: _*) + + /** + * Returns type information for [[org.apache.flink.types.Row]] with fields of the given types + * and with given names. A row must not be null. + * + * A row is a variable-length, null-aware composite type for storing multiple values in a + * deterministic field order. Every field can be null independent of the field's type. + * The type of row fields cannot be automatically inferred; therefore, it is required to pass + * type information whenever a row is used. + * + * <p>The schema of rows can have up to <code>Integer.MAX_VALUE</code> fields, however, all row instances + * must have the same length otherwise serialization fails or information is lost. + * + * Example use: `Types.ROW(Array("name", "number"), Array(Types.STRING, Types.INT))`. + * + * @param fieldNames array of field names + * @param types array of field types + */ + def ROW(fieldNames: Array[String], types: Array[TypeInformation[_]]): TypeInformation[Row] = + JTypes.ROW_NAMED(fieldNames, types: _*) + + /** + * Returns type information for a POJO (Plain Old Java Object). + * + * A POJO class is public and standalone (no non-static inner class). It has a public + * no-argument constructor. All non-static, non-transient fields in the class (and all + * superclasses) are either public (and non-final) or have a public getter and a setter + * method that follows the Java beans naming conventions for getters and setters. + * + * A POJO is a fixed-length, null-aware composite type with non-deterministic field order. + * Every field can be null independent of the field's type. + * + * The generic types for all fields of the POJO can be defined in a hierarchy of subclasses. + * + * If Flink's type analyzer is unable to extract a valid POJO type information with + * type information for all fields, an + * [[org.apache.flink.api.common.functions.InvalidTypesException}]] is thrown. + * Alternatively, you can use [[Types.POJO(Class, Map)]] to specify all fields manually. + * + * @param pojoClass POJO class to be analyzed by Flink + */ + def POJO[T](pojoClass: Class[T]): TypeInformation[T] = { + JTypes.POJO(pojoClass) + } + + /** + * Returns type information for a POJO (Plain Old Java Object) and allows to specify all + * fields manually. + * + * A POJO class is public and standalone (no non-static inner class). It has a public no-argument + * constructor. All non-static, non-transient fields in the class (and all superclasses) are + * either public (and non-final) or have a public getter and a setter method that follows the + * Java beans naming conventions for getters and setters. + * + * A POJO is a fixed-length, null-aware composite type with non-deterministic field order. + * Every field can be null independent of the field's type. + * + * The generic types for all fields of the POJO can be defined in a hierarchy of subclasses. + * + * If Flink's type analyzer is unable to extract a POJO field, an + * [[org.apache.flink.api.common.functions.InvalidTypesException]] is thrown. + * + * '''Note:''' In most cases the type information of fields can be determined automatically, + * we recommend to use [[Types.POJO(Class)]]. + * + * @param pojoClass POJO class + * @param fields map of fields that map a name to type information + */ + def POJO[T](pojoClass: Class[T], fields: Map[String, TypeInformation[_]]): TypeInformation[T] = { + JTypes.POJO(pojoClass, fields.asJava) + } + + /** + * Returns generic type information for any Scala/Java object. The serialization logic will + * use the general purpose serializer Kryo. + * + * Generic types are black-boxes for Flink, but allow any object and null values in fields. + * + * By default, serialization of this type is not very efficient. Please read the documentation + * about how to improve efficiency (namely by pre-registering classes). + * + * @param genericClass any Scala/Java class + */ + def GENERIC[T](genericClass: Class[T]): TypeInformation[T] = JTypes.GENERIC(genericClass) + + /** + * Returns type information for a Scala case class and Scala tuples. + * + * A Scala case class is a fixed-length composite type for storing multiple values in a + * deterministic field order. Fields of a case class are typed. Case classes and tuples are + * the most efficient composite type; therefore, they do not support null values unless + * its field type supports nullability. + * + * Example use: `Types.CASE_CLASS[MyCaseClass]` + * + * @tparam T case class to be analyzed + */ + def CASE_CLASS[T: TypeInformation]: TypeInformation[T] = { --- End diff -- I think an alias `TUPLE` of this method would be good. > Add helper methods for all built-in Flink types to Types > -------------------------------------------------------- > > Key: FLINK-7452 > URL: https://issues.apache.org/jira/browse/FLINK-7452 > Project: Flink > Issue Type: Improvement > Components: Type Serialization System > Reporter: Timo Walther > Assignee: Timo Walther > > Sometimes it is very difficult to provide `TypeInformation` manually, in case > some extraction fails or is not available. {{TypeHint}}s should be the > preferred way but this methods can ensure correct types. > I propose to add all built-in Flink types to the {{Types}}. Such as: > {code} > Types.POJO(MyPojo.class) > Types.POJO(Map<String, TypeInformation>) > Types.GENERIC(Object.class) > Types.TUPLE(TypeInformation, ...) > Types.MAP(TypeInformation, TypeInformation) > {code} > The methods should validate that the returned type is exactly the requested > type. And especially in case of POJO should help creating {{PojoTypeInfo}}. > Once this is in place, we can deprecate the {{TypeInfoParser}}. -- This message was sent by Atlassian JIRA (v6.4.14#64029)