Data Types

Due to historical reasons, before Flink 1.9, Flink’s Table & SQL API data types were tightly coupled to Flink’s TypeInformation. TypeInformation is used in the DataStream and DataSet API and is sufficient to describe all information needed to serialize and deserialize JVM-based objects in a distributed setting.

However, TypeInformation was not designed to represent logical types independent of an actual JVM class. In the past, it was difficult to map SQL standard types to this abstraction. Furthermore, some types were not SQL-compliant and introduced without a bigger picture in mind.

Attention Please see the planner compatibility table and limitations section before using a data type.

Users of the JVM-based API work with instances of org.apache.flink.table.types.DataType within the Table API or when defining connectors, catalogs, or user-defined functions.

A DataType instance has two responsibilities:

• Declaration of a logical type which does not imply a concrete physical representation for transmission or storage but defines the boundaries between JVM-based/Python languages and the table ecosystem.
• Optional: Giving hints about the physical representation of data to the planner which is useful at the edges to other APIs.

For JVM-based languages, all pre-defined data types are available in org.apache.flink.table.api.DataTypes.

Users of the Python API work with instances of pyflink.table.types.DataType within the Python Table API or when defining Python user-defined functions.

A DataType instance has such a responsibility:

• Declaration of a logical type which does not imply a concrete physical representation for transmission or storage but defines the boundaries between Python languages and the table ecosystem.

For Python language, those types are available in pyflink.table.types.DataTypes.

It is recommended to add a star import to your table programs for having a fluent API:

import static org.apache.flink.table.api.DataTypes.*;

DataType t = INTERVAL(DAY(), SECOND(3));

It is recommended to add a star import to your table programs for having a fluent API:

import org.apache.flink.table.api.DataTypes._

val t: DataType = INTERVAL(DAY(), SECOND(3));

Physical Hints

Physical hints are required at the edges of the table ecosystem where the SQL-based type system ends and programming-specific data types are required. Hints indicate the data format that an implementation expects.

For example, a data source could express that it produces values for logical TIMESTAMPs using a java.sql.Timestamp class instead of using java.time.LocalDateTime which would be the default. With this information, the runtime is able to convert the produced class into its internal data format. In return, a data sink can declare the data format it consumes from the runtime.

Here are some examples of how to declare a bridging conversion class:

// tell the runtime to not produce or consume java.time.LocalDateTime instances
// but java.sql.Timestamp
DataType t = DataTypes.TIMESTAMP(3).bridgedTo(java.sql.Timestamp.class);

// tell the runtime to not produce or consume boxed integer arrays
// but primitive int arrays
DataType t = DataTypes.ARRAY(DataTypes.INT().notNull()).bridgedTo(int[].class);

// tell the runtime to not produce or consume java.time.LocalDateTime instances
// but java.sql.Timestamp
val t: DataType = DataTypes.TIMESTAMP(3).bridgedTo(classOf[java.sql.Timestamp]);

// tell the runtime to not produce or consume boxed integer arrays
// but primitive int arrays
val t: DataType = DataTypes.ARRAY(DataTypes.INT().notNull()).bridgedTo(classOf[Array[Int]]);

Attention Please note that physical hints are usually only required if the API is extended. Users of predefined sources/sinks/functions do not need to define such hints. Hints within a table program (e.g. field.cast(TIMESTAMP(3).bridgedTo(Timestamp.class))) are ignored.

As mentioned in the introduction, reworking the type system will span multiple releases, and the support of each data type depends on the used planner. This section aims to summarize the most significant differences.

This part is for Java/Scala users. There are no known similiar planner compatibility issues on the data types of Python Table API currently.

Flink’s old planner, introduced before Flink 1.9, primarily supports type information. It has only limited support for data types. It is possible to declare data types that can be translated into type information such that the old planner understands them.

The following table summarizes the difference between data type and type information. Most simple types, as well as the row type remain the same. Time types, array types, and the decimal type need special attention. Other hints as the ones mentioned are not allowed.

For the Type Information column the table omits the prefix org.apache.flink.table.api.Types.

For the Data Type Representation column the table omits the prefix org.apache.flink.table.api.DataTypes.

Attention If there is a problem with the new type system. Users can fallback to type information defined in org.apache.flink.table.api.Types at any time.

N/A

The new Blink planner supports all of types of the old planner. This includes in particular the listed Java expression strings and type information.

The following data types are supported:

N/A

Java Expression String: Java expression strings in the Table API such as table.select("field.cast(STRING)") have not been updated to the new type system yet. Use the string representations declared in the old planner section.

User-defined Functions: User-defined aggregate functions cannot declare a data type yet. Scalar and table functions fully support data types.

This part is for Java/Scala users. There are no known similiar limitations on the data types of Python Table API currently.

For the JVM-based Table API those types are also available in org.apache.flink.table.api.DataTypes.

For the Python Table API, those types are available in pyflink.table.types.DataTypes.

DataTypes.CHAR(n)

Bridging to JVM Types

The type can be declared using CHAR(n) where n is the number of code points. n must have a value between 1 and 2,147,483,647 (both inclusive). If no length is specified, n is equal to 1.

DataTypes.VARCHAR(n)

DataTypes.STRING()

Bridging to JVM Types

DataTypes.VARCHAR(n)

DataTypes.STRING()

Attention The specified maximum number of code points n in DataTypes.VARCHAR(n) must be 2,147,483,647 currently.

DataTypes.BINARY(n)

Bridging to JVM Types

The type can be declared using BINARY(n) where n is the number of bytes. n must have a value between 1 and 2,147,483,647 (both inclusive). If no length is specified, n is equal to 1.

DataTypes.VARBINARY(n)

DataTypes.BYTES()

Bridging to JVM Types

DataTypes.VARBINARY(n)

DataTypes.BYTES()

Attention The specified maximum number of bytes n in DataTypes.VARBINARY(n) must be 2,147,483,647 currently.

DataTypes.DECIMAL(p, s)

Bridging to JVM Types

DataTypes.DECIMAL(p, s)

Attention The precision and scale specified in DataTypes.DECIMAL(p, s) must be 38 and 18 separately currently.

DataTypes.TINYINT()

Bridging to JVM Types

DataTypes.SMALLINT()

Bridging to JVM Types

DataTypes.INT()

Bridging to JVM Types

DataTypes.BIGINT()

Bridging to JVM Types

DataTypes.FLOAT()

Bridging to JVM Types

DataTypes.DOUBLE()

Bridging to JVM Types

DataTypes.DATE()

Bridging to JVM Types

Compared to the SQL standard, leap seconds (23:59:60 and 23:59:61) are not supported as the semantics are closer to java.time.LocalTime. A time with time zone is not provided.

Compared to the SQL standard, leap seconds (23:59:60 and 23:59:61) are not supported. A time with time zone is not provided.

DataTypes.TIME(p)

Bridging to JVM Types

DataTypes.TIME(p)

Attention The precision specified in DataTypes.TIME(p) must be 0 currently.

Compared to the SQL standard, leap seconds (23:59:60 and 23:59:61) are not supported as the semantics are closer to java.time.LocalDateTime.

A conversion from and to BIGINT (a JVM long type) is not supported as this would imply a time zone. However, this type is time zone free. For more java.time.Instant-like semantics use TIMESTAMP WITH LOCAL TIME ZONE.

Compared to the SQL standard, leap seconds (23:59:60 and 23:59:61) are not supported.

A conversion from and to BIGINT is not supported as this would imply a time zone. However, this type is time zone free. If you have such a requirement please use TIMESTAMP WITH LOCAL TIME ZONE.

DataTypes.TIMESTAMP(p)

Bridging to JVM Types

DataTypes.TIMESTAMP(p)

Attention The precision specified in DataTypes.TIMESTAMP(p) must be 3 currently.

Compared to the SQL standard, leap seconds (23:59:60 and 23:59:61) are not supported as the semantics are closer to java.time.OffsetDateTime.

Compared to the SQL standard, leap seconds (23:59:60 and 23:59:61) are not supported.

DataTypes.TIMESTAMP_WITH_TIME_ZONE(p)

Bridging to JVM Types

The type can be declared using TIMESTAMP(p) WITH TIME ZONE where p is the number of digits of fractional seconds (precision). p must have a value between 0 and 9 (both inclusive). If no precision is specified, p is equal to 6.

Leap seconds (23:59:60 and 23:59:61) are not supported as the semantics are closer to java.time.OffsetDateTime.

Compared to TIMESTAMP WITH TIME ZONE, the time zone offset information is not stored physically in every datum. Instead, the type assumes java.time.Instant semantics in UTC time zone at the edges of the table ecosystem. Every datum is interpreted in the local time zone configured in the current session for computation and visualization.

Leap seconds (23:59:60 and 23:59:61) are not supported.

Compared to TIMESTAMP WITH TIME ZONE, the time zone offset information is not stored physically in every datum. Every datum is interpreted in the local time zone configured in the current session for computation and visualization.

DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(p)

Bridging to JVM Types

DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(p)

Attention The precision specified in DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(p) must be 3 currently.

DataTypes.INTERVAL(DataTypes.YEAR())
DataTypes.INTERVAL(DataTypes.YEAR(p))
DataTypes.INTERVAL(DataTypes.YEAR(p), DataTypes.MONTH())
DataTypes.INTERVAL(DataTypes.MONTH())

Bridging to JVM Types

DataTypes.INTERVAL(DataTypes.DAY())
DataTypes.INTERVAL(DataTypes.DAY(p1))
DataTypes.INTERVAL(DataTypes.DAY(p1), DataTypes.HOUR())
DataTypes.INTERVAL(DataTypes.DAY(p1), DataTypes.MINUTE())
DataTypes.INTERVAL(DataTypes.DAY(p1), DataTypes.SECOND(p2))
DataTypes.INTERVAL(DataTypes.HOUR())
DataTypes.INTERVAL(DataTypes.HOUR(), DataTypes.MINUTE())
DataTypes.INTERVAL(DataTypes.HOUR(), DataTypes.SECOND(p2))
DataTypes.INTERVAL(DataTypes.MINUTE())
DataTypes.INTERVAL(DataTypes.MINUTE(), DataTypes.SECOND(p2))
DataTypes.INTERVAL(DataTypes.SECOND())
DataTypes.INTERVAL(DataTypes.SECOND(p2))

Bridging to JVM Types

DataTypes.ARRAY(t)

Bridging to JVM Types

DataTypes.MAP(kt, vt)

Bridging to JVM Types

DataTypes.MULTISET(t)

Bridging to JVM Types

DataTypes.ROW(DataTypes.FIELD(n0, t0), DataTypes.FIELD(n1, t1), ...)
DataTypes.ROW(DataTypes.FIELD(n0, t0, d0), DataTypes.FIELD(n1, t1, d1), ...)

Bridging to JVM Types

Attention User-defined data types are not fully supported yet. They are currently (as of Flink 1.11) only exposed as unregistered structured types in parameters and return types of functions.

A structured type is similar to an object in an object-oriented programming language. It contains zero, one or more attributes. Each attribute consists of a name and a type.

There are two kinds of structured types:

• Types that are stored in a catalog and are identified by a catalog identifer (like cat.db.MyType). Those are equal to the SQL standard definition of structured types.

• Anonymously defined, unregistered types (usually reflectively extracted) that are identified by an implementation class (like com.myorg.model.MyType). Those are useful when programmatically defining a table program. They enable reusing existing JVM classes without manually defining the schema of a data type again.

Registered Structured Types

Currently, registered structured types are not supported. Thus, they cannot be stored in a catalog or referenced in a CREATE TABLE DDL.

Unregistered Structured Types

Unregistered structured types can be created from regular POJOs (Plain Old Java Objects) using automatic reflective extraction.

The implementation class of a structured type must meet the following requirements:

• The class must be globally accessible which means it must be declared public, static, and not abstract.
• The class must offer a default constructor with zero arguments or a full constructor that assigns all fields.
• All fields of the class must be readable by either public declaration or a getter that follows common coding style such as getField(), isField(), field().
• All fields of the class must be writable by either public declaration, fully assigning constructor, or a setter that follows common coding style such as setField(...), field(...).
• All fields must be mapped to a data type either implicitly via reflective extraction or explicitly using the @DataTypeHint annotations.
• Fields that are declared static or transient are ignored.

The reflective extraction supports arbitrary nesting of fields as long as a field type does not (transitively) refer to itself.

The declared field class (e.g. public int age;) must be contained in the list of supported JVM bridging classes defined for every data type in this document (e.g. java.lang.Integer or int for INT).

For some classes an annotation is required in order to map the class to a data type (e.g. @DataTypeHint("DECIMAL(10, 2)") to assign a fixed precision and scale for java.math.BigDecimal).

class User {

    // extract fields automatically
    public int age;
    public String name;

    // enrich the extraction with precision information
    public @DataTypeHint("DECIMAL(10, 2)") BigDecimal totalBalance;

    // enrich the extraction with forcing using RAW types
    public @DataTypeHint("RAW") Class<?> modelClass;
}

DataTypes.of(User.class);

Bridging to JVM Types

case class User(

    // extract fields automatically
    age: Int,
    name: String,

    // enrich the extraction with precision information
    @DataTypeHint("DECIMAL(10, 2)") totalBalance: java.math.BigDecimal,

    // enrich the extraction with forcing using a RAW type
    @DataTypeHint("RAW") modelClass: Class[_]
)

DataTypes.of(classOf[User])

Bridging to JVM Types

DataTypes.BOOLEAN()

Bridging to JVM Types

DataTypes.RAW(class, serializer)

DataTypes.RAW(class)

Bridging to JVM Types

The type can be declared using RAW('class', 'snapshot') where class is the originating class and snapshot is the serialized TypeSerializerSnapshot in Base64 encoding. Usually, the type string is not declared directly but is generated while persisting the type.

In the API, the RAW type can be declared either by directly supplying a Class + TypeSerializer or by passing Class and letting the framework extract Class + TypeSerializer from there.

DataTypes.NULL()

Bridging to JVM Types

At many locations in the API, Flink tries to automatically extract data type from class information using reflection to avoid repetitive manual schema work. However, extracting a data type reflectively is not always successful because logical information might be missing. Therefore, it might be necessary to add additional information close to a class or field declaration for supporting the extraction logic.

The following table lists classes that can be implicitly mapped to a data type without requiring further information.

If you intend to implement classes in Scala, it is recommended to use boxed types (e.g. java.lang.Integer) instead of Scala’s primitives. Scala’s primitives (e.g. Int or Double) are compiled to JVM primitives (e.g. int/double) and result in NOT NULL semantics as shown in the table below. Furthermore, Scala primitives that are used in generics (e.g. java.util.Map[Int, Double]) are erased during compilation and lead to class information similar to java.util.Map[java.lang.Object, java.lang.Object].

Other JVM bridging classes mentioned in this document require a @DataTypeHint annotation.

Data type hints can parameterize or replace the default extraction logic of individual function parameters and return types, structured classes, or fields of structured classes. An implementer can choose to what extent the default extraction logic should be modified by declaring a @DataTypeHint annotation.

The @DataTypeHint annotation provides a set of optional hint parameters. Some of those parameters are shown in the following example. More information can be found in the documentation of the annotation class.