# Data Types in Confluent Cloud for Apache Flink Confluent Cloud for Apache Flink® has a rich set of native data types that you can use in SQL statements and queries. The query planner supports the Flink SQL types that are described in this topic. The following table shows the mapping of Flink SQL types to Java and Python types. For the mapping of Flink SQL types to JSON Schema, Protobuf, and Avro types, see [Data Type Mappings for serialization](serialization.md#flink-sql-serialization). | Flink SQL type | Java type | Python type | |------------------------------------------------------|----------------------------|-----------------------------| | [ARRAY](#flink-sql-array) | t[] | numpy.ndarray, list[t] | | [BIGINT](#flink-sql-bigint) | long | int | | [BINARY](#flink-sql-binary) | byte[] | bytes, bytearray | | [BOOLEAN](#flink-sql-boolean) | boolean | bool() | | [BYTES / VARBINARY](#flink-sql-bytes) | byte[] | bytes, bytearray | | [CHAR](#flink-sql-char) | String | str | | [DATE](#flink-sql-date) | java.time.LocalDate | datetime.date | | [DECIMAL](#flink-sql-decimal) | java.math.BigDecimal | decimal.Decimal | | [DOUBLE](#flink-sql-double) | double | float | | [FLOAT](#flink-sql-float) | float | float | | [INT](#flink-sql-int) | long | int | | [INTERVAL DAY TO SECOND](#flink-sql-interval-d-to-s) | java.time.Duration | datetime.timedelta | | [INTERVAL YEAR TO MONTH](#flink-sql-interval-y-to-m) | java.time.Period | dateutil.relativedelta | | [MAP](#flink-sql-map) | java.util.Map | dict[kt, vt] | | [MULTISET](#flink-sql-multiset) | java.util.Map | dict[t, int] | | [NULL](#flink-sql-null) | java.lang.Object | object | | [ROW](#flink-sql-row) | org.apache.flink.types.Row | pyflink.table.types.RowType | | [SMALLINT](#flink-sql-smallint) | short | int | | [TIME](#flink-sql-time) | java.time.LocalTime | datetime.time | | [TIMESTAMP](#flink-sql-timestamp) | java.time.LocalDateTime | datetime.datetime | | [TIMESTAMP_LTZ](#flink-sql-timestamp-ltz) | java.time.Instant | datetime.datetime | | [TINYINT](#flink-sql-tinyint) | byte | bytes, bytearray | | [VARCHAR / STRING](#flink-sql-varchar) | String | str | ## Data type definition A *data type* describes the logical type of a value in a SQL table. You use data types to declare the input and output types of an operation. The Flink data types are similar to the SQL standard data type terminology, but for efficient handling of scalar expressions, they also contain information about the nullability of a value. These are examples of SQL data types: ```sql INT INT NOT NULL INTERVAL DAY TO SECOND(3) ROW, fieldTwo TIMESTAMP(3)> ``` The following sections list all pre-defined data types in Flink SQL. ## Character strings ### CHAR Represents a fixed-length character string. ### SQL **Syntax** ```sql CHAR CHAR(n) ``` **Example** ```sql CREATE TABLE test_sql_char (val CHAR(3)); INSERT INTO test_sql_char VALUES ('ABCD'); -- returns 'ABC' SELECT * FROM test_sql_char; ``` Declare this type by 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*. `CHAR(0)` is not supported for CAST or persistence in catalogs, but it exists in protocols. ### Java **Syntax** ```java char ``` **Example** ```java char charValue1 = 'A'; char charValue2 = 65; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |----------------------------------------|---------|----------|-------------------------| | java.lang.String | ✓ | ✓ | Default | | byte[] | ✓ | ✓ | Assumes UTF-8 encoding | | org.apache.flink.table.data.StringData | ✓ | ✓ | Internal data structure | ### Python Not supported. Python has no `char` type, so a single character is treated as a string with length 1. **Formats** The following table shows examples of the CHAR type in different formats. | JSON for data type | ```json
{"type":"CHAR","nullable":true,"length":8}
``` | |---------------------------|------------------------------------------------------------------| | CLI/UI format | ```sql
CHAR(8)
``` | | JSON for payload | ```none
"Example string"
``` | | CLI/UI format for payload | ```none
Example string
``` | | Statement literal | ```sql
CAST('ABC' AS CHAR(3))
``` | ### VARCHAR / STRING Represents a variable-length character string. ### SQL **Syntax** ```sql VARCHAR VARCHAR(n) STRING ``` **Example** ```sql CREATE TABLE test_sql_varchar (val VARCHAR(3)); INSERT INTO test_sql_varchar VALUES ('ABCDEF'); -- returns 'ABC' SELECT * FROM test_sql_varchar; ``` Declare this type by using `VARCHAR(n)`, where `n` is the maximum 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*. `STRING` is equivalent to `VARCHAR(2147483647)`. `VARCHAR(0)` is not supported for CAST or persistence in catalogs, but it exists in protocols. ### Java **Syntax** ```java string ``` **Example** ```java string stringValue = "Hello World"; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |----------------------------------------|---------|----------|-------------------------| | java.lang.String | ✓ | ✓ | Default | | byte[] | ✓ | ✓ | Assumes UTF-8 encoding | | org.apache.flink.table.data.StringData | ✓ | ✓ | Internal data structure | ### Python **Examples** ```python stringValue = "Hello World" number = 123 stringValue1 = str(number) # Converts integer to string ``` **Formats** The following table shows examples of the VARCHAR type in different formats. | JSON for data type | ```json
{"type":"VARCHAR","nullable":true,"length":8}
``` | |---------------------------|---------------------------------------------------------------------| | CLI/UI format | ```sql
VARCHAR(800)
``` | | JSON for payload | ```none
"Example string"
``` | | CLI/UI format for payload | ```none
Example string
``` | | Statement literal | ```sql
'Hello world'
``` | ## Binary strings ### BINARY Represents a fixed-length binary string, or a sequence of bytes. ### SQL **Syntax** ```sql BINARY BINARY(n) ``` **Example** ```sql CREATE TABLE test_sql_binary (val BINARY(3)); INSERT INTO test_sql_binary VALUES (X'7f0203'); -- returns x'7f0203' SELECT * FROM test_sql_binary; ``` Declare this type by 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*. The string representation is hexadecimal format. `BINARY(0)` is not supported for CAST or persistence in catalogs, but it exists in protocols. ### Java **Syntax** ```java byte[] ``` **Example** ```java byte[] binaryValue = {0x7f, 0x02, 0x03}; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------|---------|----------|---------| | byte[] | ✓ | ✓ | Default | ### Python **Example** ```python data = bytes([65, 66, 67]) ``` **Formats** The following table shows examples of the BINARY type in different formats. | JSON for data type | ```json
{"type":"BINARY","nullable":true,"length":1}
``` | |---------------------------|--------------------------------------------------------------------| | CLI/UI format | ```sql
BINARY(3)
``` | | JSON for payload | ```none
"x'7f0203'"
``` | | CLI/UI format for payload | ```none
x'7f0203'
``` | | Statement literal | ```sql
CAST(X'7f0203' AS BINARY(3))
``` | ### BYTES / VARBINARY Represents a variable-length binary string (=a sequence of bytes). ### SQL **Syntax** ```sql BYTES VARBINARY VARBINARY(n) ``` **Example** ```sql CREATE TABLE test_sql_bytes (val VARBINARY(3)); INSERT INTO test_sql_bytes VALUES (X'7f0203040506'); -- returns x'7f0203' SELECT * FROM test_sql_bytes; ``` Declare this type by using `VARBINARY(n)` where `n` is the maximum 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*. `BYTES` is equivalent to `VARBINARY(2147483647)`. `VARCHAR(0)` is not supported for CAST or persistence in catalogs, but it exists in protocols. ### Java **Syntax** ```java byte[] ``` **Example** ```java byte[] binaryValue = {0x7f, 0x02, 0x03}; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------|---------|----------|---------| | byte[] | ✓ | ✓ | Default | ### Python **Syntax** ```python bytes ``` **Example** ```python bytesValue = bytes([65, 66, 67]) ``` **Formats** The following table shows examples of the VARBINARY type in different formats. | JSON for data type | ```json
{"type":"VARBINARY","nullable":true,"length":1}
``` | |---------------------------|-----------------------------------------------------------------------| | CLI/UI format | ```sql
VARBINARY(800)
``` | | JSON for payload | ```none
"x'7f0203'"
``` | | CLI/UI format for payload | ```none
x'7f0203'
``` | | Statement literal | ```sql
X'7f0203'
``` | ## Exact numerics ### BIGINT Represents an 8-byte signed integer with values from *-9,223,372,036,854,775,808* to *9,223,372,036,854,775,807*. ### SQL **Syntax** ```sql BIGINT ``` **Example** ```sql CREATE TABLE test_sql_bigint (val BIGINT); INSERT INTO test_sql_bigint VALUES (1234567890123456789); -- returns 1234567890123456800 SELECT * FROM test_sql_bigint; ``` ### Java **Syntax** ```java long ``` **Example** ```java long longValue = 1234567890123456789; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |----------------|---------|----------|-------------------------------------| | java.lang.Long | ✓ | ✓ | Default | | long | ✓ | (✓) | Output only if type is not nullable | ### Python **Syntax** ```python int ``` **Example** ```python longValue = 1234567890123456789 ``` **Formats** The following table shows examples of the BIGINT type in different formats. | JSON for data type | ```json
{"type":"BIGINT","nullable":true}
``` | |---------------------------|---------------------------------------------------------| | CLI/UI format | ```sql
BIGINT
``` | | JSON for payload | ```none
"23"
``` | | CLI/UI format for payload | ```none
23
``` | | Statement literal | ```sql
CAST(23 AS BIGINT)
``` | ### DECIMAL Represents a decimal number with fixed precision and scale. ### SQL **Syntax** ```sql DEC DEC(p) DEC(p, s) DECIMAL DECIMAL(p) DECIMAL(p, s) NUMERIC NUMERIC(p) NUMERIC(p, s) ``` **Example** ```sql CREATE TABLE test_sql_decimal (val DECIMAL(5, 3)); INSERT INTO test_sql_decimal VALUES (12.123), (123.12345); -- returns 12.123, NULL SELECT * FROM test_sql_decimal; ``` Declare this type by using `DECIMAL(p, s)` where `p` is the number of digits in a number (*precision*) and `s` is the number of digits to the right of the decimal point in a number (*scale*). `p` must have a value between *1* and *38* (both inclusive). The default value for `p` is *10*. `s` must have a value between *0* and `p` (both inclusive). The default value for `s` is *0*. The trailing side is padded with *0*. The leading side must be padded with spaces, as with all other values. `NUMERIC(p, s)` and `DEC(p, s)` are synonyms for this type. ### Java **Syntax** ```java BigDecimal ``` **Example** ```java BigDecimal decimalValue = new BigDecimal("12.123"); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-----------------------------------------|---------|----------|-------------------------| | java.math.BigDecimal | ✓ | ✓ | Default | | org.apache.flink.table.data.DecimalData | ✓ | ✓ | Internal data structure | ### Python **Syntax** ```python decimal.Decimal ``` **Example** ```python decimalValue = decimal.Decimal("12.123") ``` **Formats** The following table shows examples of the DECIMAL type in different formats. | JSON for data type | ```json
{"type":"DECIMAL","nullable":true,"precision":5,"scale":3}
``` | |---------------------------|----------------------------------------------------------------------------------| | CLI/UI format | ```sql
DECIMAL(5, 3)
``` | | JSON for payload | ```none
"12.123"
``` | | CLI/UI format for payload | ```none
12.123
``` | | Statement literal | ```sql
12.123
``` | ### INT Represents a 4-byte signed integer with values from *-2,147,483,648* to *2,147,483,647*. ### SQL **Syntax** ```sql INT INTEGER ``` INTEGER is a synonym for this type. **Example** ```sql CREATE TABLE test_sql_int (val INT); INSERT INTO test_sql_int VALUES (1234567890); -- returns 1234567890 SELECT * FROM test_sql_int; ``` ### Java **Syntax** ```java int ``` **Example** ```java int intValue = 1234567890; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------|---------|----------|-------------------------------------| | java.lang.Integer | ✓ | ✓ | Default | | long | ✓ | (✓) | Output only if type is not nullable | ### Python **Syntax** ```python int ``` **Example** ```python intValue = 1234567890 ``` **Formats** The following table shows examples of the INT type in different formats. | JSON for data type | ```json
{"type":"INT","nullable":true}
``` | |---------------------------|------------------------------------------------------| | CLI/UI format | ```sql
INT
``` | | JSON for payload | ```none
"23"
``` | | CLI/UI format for payload | ```none
23
``` | | Statement literal | ```sql
1234567890
``` | ### SMALLINT Represents a 2-byte signed integer with values from *-32,768* to *32,767*. ### SQL **Syntax** ```sql SMALLINT ``` **Example** ```sql CREATE TABLE test_sql_smallint (val SMALLINT); INSERT INTO test_sql_smallint VALUES (CAST(1234567890 AS SMALLINT)); -- returns 722 SELECT * FROM test_sql_smallint; ``` ### Java **Syntax** ```java short ``` **Example** ```java short shortValue = 1234567890; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-----------------|---------|----------|-------------------------------------| | java.lang.Short | ✓ | ✓ | Default | | short | ✓ | (✓ ) | Output only if type is not nullable | ### Python **Syntax** ```python int ``` **Example** ```python shortValue = 1234567890 ``` **Formats** The following table shows examples of the SMALLINT type in different formats. | JSON for data type | ```json
{"type":"SMALLINT","nullable":true}
``` | |---------------------------|-----------------------------------------------------------| | CLI/UI format | ```sql
SMALLINT
``` | | JSON for payload | ```none
"23"
``` | | CLI/UI format for payload | ```none
23
``` | | Statement literal | ```sql
CAST(722 AS SMALLINT)
``` | ### TINYINT Represents a 1-byte signed integer with values from *-128* to *127*. ### SQL **Syntax** ```sql TINYINT ``` **Example** ```sql CREATE TABLE test_sql_tinyint (val TINYINT); INSERT INTO test_sql_tinyint VALUES (CAST(1234567890 AS TINYINT)); -- returns -46 SELECT * FROM test_sql_tinyint; ``` ### Java **Syntax** ```java byte ``` **Example** ```java byte byteValue = 123; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |----------------|---------|----------|-------------------------------------| | java.lang.Byte | ✓ | ✓ | Default | | byte | ✓ | (✓) | Output only if type is not nullable | ### Python **Syntax** ```python byte ``` **Example** ```python byteValue = 123 ``` **Formats** The following table shows examples of the TINYINT type in different formats. | JSON for data type | ```json
{"type":"TINYINT","nullable":true}
``` | |---------------------------|----------------------------------------------------------| | CLI/UI format | ```sql
TINYINT
``` | | JSON for payload | ```none
"23"
``` | | CLI/UI format for payload | ```none
23
``` | | Statement literal | ```sql
CAST(-46 AS TINYINT)
``` | ## Approximate numerics ### DOUBLE Represents an 8-byte double precision floating point number. ### SQL **Syntax** ```sql DOUBLE DOUBLE PRECISION ``` **Example** ```sql CREATE TABLE test_sql_double (val DOUBLE); INSERT INTO test_sql_double VALUES (1.1111112120000001E7); -- returns 11111112.120000001 SELECT * FROM test_sql_double; ``` DOUBLE PRECISION is a synonym for this type. ### Java **Syntax** ```java double ``` **Example** ```java double doubleValue = 1.1111112120000001E7; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |------------------|---------|----------|-------------------------------------| | java.lang.Double | ✓ | ✓ | Default | | double | ✓ | (✓) | Output only if type is not nullable | ### Python **Syntax** ```python float ``` **Example** ```python doubleValue = 1.1111112120000001E7 ``` **Formats** The following table shows examples of the DOUBLE type in different formats. | JSON for data type | ```json
{"type":"DOUBLE","nullable":true}
``` | |---------------------------|---------------------------------------------------------| | CLI/UI format | ```sql
DOUBLE
``` | | JSON for payload | ```none
"1.1111112120000001E7"
``` | | CLI/UI format for payload | ```none
1.1111112120000001E7
``` | | Statement literal | ```sql
1.1111112120000001E7
``` | ### FLOAT Represents a 4-byte single precision floating point number. ### SQL **Syntax** ```sql FLOAT ``` **Example** ```sql CREATE TABLE test_sql_float (val FLOAT); INSERT INTO test_sql_float VALUES (1.1111112); -- returns 1.1111112 SELECT * FROM test_sql_float; ``` Compared to the SQL standard, this type doesn’t take parameters. ### Java **Syntax** ```java float ``` **Example** ```java float floatValue = 1.1111112; ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-----------------|---------|----------|-------------------------------------| | java.lang.Float | ✓ | ✓ | Default | | float | ✓ | (✓) | Output only if type is not nullable | ### Python **Syntax** ```python float ``` **Example** ```python floatValue = 1.1111112 ``` **Formats** The following table shows examples of the FLOAT type in different formats. | JSON for data type | ```json
{"type":"FLOAT","nullable":true}
``` | |---------------------------|--------------------------------------------------------| | CLI/UI format | ```sql
FLOAT
``` | | JSON for payload | ```none
"1.1111112E7"
``` | | CLI/UI format for payload | ```none
1.1111112E7
``` | | Statement literal | ```sql
CAST(1.1111112 AS FLOAT)
``` | ## Date and time ### DATE Represents a date consisting of `year-month-day` with values ranging from `0000-01-01` to `9999-12-31`. ### SQL **Syntax** ```sql DATE ``` **Example** ```sql CREATE TABLE test_sql_date (val DATE); INSERT INTO test_sql_date VALUES (CAST('2023-04-06' AS DATE)); -- returns 2023-04-06 SELECT * FROM test_sql_date; ``` Compared to the SQL standard, the range starts at year `0000`. ### Java **Syntax** ```java date ``` **Example** ```java date dateValue = date(2023, 4, 6); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |---------------------|---------|----------|-------------------------------------------------------------------------------------| | java.time.LocalDate | ✓ | ✓ | Default | | java.sql.Date | ✓ | ✓ | | | java.lang.Integer | ✓ | ✓ | Describes the number of days since Unix epoch | | int | ✓ | (✓) | Describes the number of days since Unix epoch. Output only if type is not nullable. | ### Python **Syntax** ```python date ``` **Example** ```python dateValue = date(2023, 4, 6) ``` **Formats** The following table shows examples of the DATE type in different formats. | JSON for data type | ```json
{"type":"DATE","nullable":true}
``` | |---------------------------|-------------------------------------------------------| | CLI/UI format | ```sql
DATE
``` | | JSON for payload | ```none
"2023-04-06"
``` | | CLI/UI format for payload | ```none
2023-04-06
``` | | Statement literal | ```sql
DATE '2023-04-06'
``` | ### INTERVAL DAY TO SECOND Data type for a group of day-time interval types. #### NOTE You can’t declare an `INTERVAL DAY TO SECOND` column in a `CREATE TABLE` statement. Use this type only in expressions, like the literal shown in the following example. ### SQL **Syntax** ```sql INTERVAL DAY INTERVAL DAY(p1) INTERVAL DAY(p1) TO HOUR INTERVAL DAY(p1) TO MINUTE INTERVAL DAY(p1) TO SECOND(p2) INTERVAL HOUR INTERVAL HOUR TO MINUTE INTERVAL HOUR TO SECOND(p2) INTERVAL MINUTE INTERVAL MINUTE TO SECOND(p2) INTERVAL SECOND INTERVAL SECOND(p2) ``` **Example** ```sql -- returns +2 07:33:20.000 SELECT INTERVAL '2 07:33:20.000' DAY TO SECOND; ``` Declare this type by using the above combinations, where `p1` is the number of digits of days (*day precision*) and `p2` is the number of digits of fractional seconds (*fractional precision*). `p1` must have a value between *1* and *6* (both inclusive). If no `p1` is specified, it is equal to *2* by default. `p2` must have a value between *0* and *9* (both inclusive). If no `p2` is specified, it is equal to *6* by default. The type must be parameterized to one of these resolutions with up to nanosecond precision: - Interval of days - Interval of days to hours - Interval of days to minutes - Interval of days to seconds - Interval of hours - Interval of hours to minutes - Interval of hours to seconds - Interval of minutes - Interval of minutes to seconds - Interval of seconds An interval of day-time consists of `+days hours:months:seconds.fractional` with values ranging from `-999999 23:59:59.999999999` to `+999999 23:59:59.999999999`. The value representation is the same for all types of resolutions. For example, an interval of seconds of *70* is always represented in an interval-of-days-to-seconds format (with default precisions): `+00 00:01:10.000000`. Formatting intervals are tricky, because they have different resolutions: - DAY - DAY_TO_HOUR - DAY_TO_MINUTE - DAY_TO_SECOND - HOUR - HOUR_TO_MINUTE - HOUR_TO_SECOND - MINUTE - MINUTE_TO_SECOND - SECOND Depending on the resolution, use: ```sql INTERVAL DAY(1) INTERVAL DAY(1) TO HOUR INTERVAL DAY(1) TO MINUTE INTERVAL DAY(1) TO SECOND(3) INTERVAL HOUR INTERVAL HOUR TO MINUTE INTERVAL HOUR TO SECOND(3) INTERVAL MINUTE INTERVAL MINUTE TO SECOND(3) INTERVAL SECOND(3) ``` ### Java **Syntax** ```java Duration DataTypes.INTERVAL ``` **Example** ```java // HH:MM:SS = 2:07:33 Duration intervalValue = Duration.ofSeconds(2 * 3600 + 7 * 60 + 33); ``` **Table API examples** ```java import static org.apache.flink.table.api.DataTypes.*; // Single field intervals DataType intervalDay = INTERVAL(DAY()); DataType intervalDayWithParam = INTERVAL(DAY(p1)); DataType intervalDayToHour = INTERVAL(DAY(p1), HOUR()); DataType intervalDayToMinute = INTERVAL(DAY(p1), MINUTE()); DataType intervalDayToSecond = INTERVAL(DAY(p1), SECOND(p2)); // Hour-based intervals DataType intervalHour = INTERVAL(HOUR()); DataType intervalHourToMinute = INTERVAL(HOUR(), MINUTE()); DataType intervalHourToSecond = INTERVAL(HOUR(), SECOND(p2)); // Minute-based intervals DataType intervalMinute = INTERVAL(MINUTE()); DataType intervalMinuteToSecond = INTERVAL(MINUTE(), SECOND(p2)); // Second-based intervals DataType intervalSecond = INTERVAL(SECOND()); DataType intervalSecondWithParam = INTERVAL(SECOND(p2)); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |--------------------|---------|----------|----------------------------------------------------------------------------| | java.time.Duration | ✓ | ✓ | Default | | java.lang.Long | ✓ | ✓ | Describes the number of milliseconds | | long | ✓ | (✓) | Describes the number of milliseconds. Output only if type is not nullable. | ### Python **Syntax** ```python from pyflink.table.types import DataTypes interval_day = DataTypes.INTERVAL(DataTypes.DAY()) ``` **Examples** ```python from pyflink.table.types import DataTypes # Single field intervals interval_day = DataTypes.INTERVAL(DataTypes.DAY()) interval_day_with_param = DataTypes.INTERVAL(DataTypes.DAY(p1)) interval_day_to_hour = DataTypes.INTERVAL(DataTypes.DAY(p1), DataTypes.HOUR()) interval_day_to_minute = DataTypes.INTERVAL(DataTypes.DAY(p1), DataTypes.MINUTE()) interval_day_to_second = DataTypes.INTERVAL(DataTypes.DAY(p1), DataTypes.SECOND(p2)) # Hour-based intervals interval_hour = DataTypes.INTERVAL(DataTypes.HOUR()) interval_hour_to_minute = DataTypes.INTERVAL(DataTypes.HOUR(), DataTypes.MINUTE()) interval_hour_to_second = DataTypes.INTERVAL(DataTypes.HOUR(), DataTypes.SECOND(p2)) # Minute-based intervals interval_minute = DataTypes.INTERVAL(DataTypes.MINUTE()) interval_minute_to_second = DataTypes.INTERVAL(DataTypes.MINUTE(), DataTypes.SECOND(p2)) # Second-based intervals interval_second = DataTypes.INTERVAL(DataTypes.SECOND()) interval_second_with_param = DataTypes.INTERVAL(DataTypes.SECOND(p2)) ``` **Formats** The following table shows examples of the INTERVAL DAY TO SECOND type in different formats. | JSON for data type | ```json
{"type":"INTERVAL_DAY_TIME","nullable":true,"precision":1,"fractionalPrecision":3,"resolution":"DAY_TO_SECOND"}
``` | |---------------------------|---------------------------------------------------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
INTERVAL DAY(1) TO SECOND(3)
``` | | JSON for payload | ```none
"+2 07:33:20.000"
``` | | CLI/UI format for payload | ```none
+2 07:33:20.000
``` | | Statement literal | ```sql
INTERVAL '2 07:33:20.000' DAY TO SECOND
``` | ### INTERVAL YEAR TO MONTH Data type for a group of year-month interval types. #### NOTE You can’t declare an `INTERVAL YEAR TO MONTH` column in a `CREATE TABLE` statement. Use this type only in expressions, like the literal shown in the following example. ### SQL **Syntax** ```sql INTERVAL YEAR INTERVAL YEAR(p) INTERVAL YEAR(p) TO MONTH INTERVAL MONTH ``` **Example** ```sql -- returns +2000-02 SELECT INTERVAL '2000-02' YEAR(4) TO MONTH; ``` Declare this type by using the above combinations, where `p` is the number of digits of years (*year precision*). `p` must have a value between *1* and *4* (both inclusive). If no year precision is specified, `p` is equal to *2*. The type must be parameterized to one of these resolutions: - Interval of years - Interval of years to months - Interval of months An interval of year-month consists of `+years-months` with values ranging from `-9999-11` to `+9999-11`. The value representation is the same for all types of resolutions. For example, an interval of months of *50* is always represented in an interval-of-years-to-months format (with default year precision): `+04-02`. Formatting intervals are tricky, because they have different resolutions: - YEAR - YEAR_TO_MONTH - MONTH Depending on the resolution, use: ```sql INTERVAL YEAR(4) INTERVAL YEAR(4) TO MONTH INTERVAL MONTH ``` ### Java **Syntax** ```java Period DataTypes.INTERVAL ``` ```java import static org.apache.flink.table.api.DataTypes.*; DataType intervalYear = INTERVAL(YEAR()); DataType intervalYearWithParam = INTERVAL(YEAR(p)); DataType intervalYearToMonth = INTERVAL(YEAR(p), MONTH()); DataType intervalMonth = INTERVAL(MONTH()); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------|---------|----------|----------------------------------------------------------------------| | java.time.Period | ✓ | ✓ | Default. Ignores the `days` part. | | java.lang.Integer | ✓ | ✓ | Describes the number of months. | | int | ✓ | (✓) | Describes the number of months. Output only if type is not nullable. | ### Python **Syntax** ```python from pyflink.table.types import DataTypes interval_year_to_month = DataTypes.INTERVAL(DataTypes.YEAR(p), DataTypes.MONTH()) ``` **Examples** ```python from pyflink.table.types import DataTypes interval_year = DataTypes.INTERVAL(DataTypes.YEAR()) interval_year_with_param = DataTypes.INTERVAL(DataTypes.YEAR(p)) interval_year_to_month = DataTypes.INTERVAL(DataTypes.YEAR(p), DataTypes.MONTH()) interval_month = DataTypes.INTERVAL(DataTypes.MONTH()) ``` **Formats** The following table shows examples of the INTERVAL YEAR TO MONTH type in different formats. | JSON for data type | ```json
{"type":"INTERVAL_YEAR_MONTH","nullable":true,"precision":4,"resolution":"YEAR_TO_MONTH"}
``` | |---------------------------|-----------------------------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
INTERVAL YEAR(4) TO MONTH
``` | | JSON for payload | ```none
"+2000-02"
``` | | CLI/UI format for payload | ```none
+2000-02
``` | | Statement literal | ```sql
INTERVAL '2000-02' YEAR(4) TO MONTH
``` | ### TIME Represents a time *without* timezone consisting of `hour:minute:second[.fractional]` with up to nanosecond precision and values ranging from `00:00:00.000000000` to `23:59:59.999999999`. ### SQL **Syntax** ```sql TIME TIME(p) TIME_WITHOUT_TIME_ZONE TIME_WITHOUT_TIME_ZONE(p) ``` **Example** ```sql CREATE TABLE test_sql_time (val TIME(3)); INSERT INTO test_sql_time VALUES (CAST('10:56:22.541' AS TIME(3))); -- returns 10:56:22.541 SELECT * FROM test_sql_time; ``` Declare this type by using `TIME(p)`, 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 *0*. 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* timezone is not provided. `TIME` acts like a pure string and isn’t related to a time zone of any kind, including UTC. `TIME WITHOUT TIME ZONE` is a synonym for this type. ### Java **Syntax** ```java Time TimeWithoutTimeZone DataTypes.TIME() DataTypes.TIME(p) DataTypes.TIME_WITHOUT_TIME_ZONE() DataTypes.TIME_WITHOUT_TIME_ZONE(p) ``` **Examples** ```java import static org.apache.flink.table.api.DataTypes.*; DataType time = TIME(); DataType timeWithPrecision = TIME(p); DataType timeWithoutTimeZone = TIME_WITHOUT_TIME_ZONE(); DataType timeWithoutTimeZoneWithPrecision = TIME_WITHOUT_TIME_ZONE(p); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |---------------------|---------|----------|---------------------------------------------------------------------------------------| | java.time.LocalTime | ✓ | ✓ | Default | | java.sql.Time | ✓ | ✓ | | | java.lang.Integer | ✓ | ✓ | Describes the number of milliseconds of the day. | | int | ✓ | (✓) | Describes the number of milliseconds of the day. Output only if type is not nullable. | | java.lang.Long | ✓ | ✓ | Describes the number of nanoseconds of the day. | | long | ✓ | (✓) | Describes the number of nanoseconds of the day. Output only if type is not nullable. | ### Python **Syntax** ```python from pyflink.table.types import DataTypes DataTypes.TIME() DataTypes.TIME(p) DataTypes.TIME_WITHOUT_TIME_ZONE() DataTypes.TIME_WITHOUT_TIME_ZONE(p) ``` **Examples** ```python from pyflink.table.types import DataTypes time_type = DataTypes.TIME() time_with_param = DataTypes.TIME(p) time_without_tz = DataTypes.TIME_WITHOUT_TIME_ZONE() time_without_tz_with_param = DataTypes.TIME_WITHOUT_TIME_ZONE(p) ``` The precision specified in `p` must be `0`. **Formats** The following table shows examples of the TIME type in different formats. | JSON for data type | ```json
{"type":"TIME_WITHOUT_TIME_ZONE","nullable":true,"precision":3}
``` | |---------------------------|---------------------------------------------------------------------------------------| | CLI/UI format | ```sql
TIME(3)
``` | | JSON for payload | ```none
"10:56:22.541"
``` | | CLI/UI format for payload | ```none
10:56:22.541
``` | | Statement literal | ```sql
TIME '10:56:22.541'
``` | ### TIMESTAMP Represents a timestamp *without* timezone consisting of `year-month-day hour:minute:second[.fractional]` with up to nanosecond precision and values ranging from `0000-01-01 00:00:00.000000000` to `9999-12-31 23:59:59.999999999`. ### SQL **Syntax** ```sql TIMESTAMP TIMESTAMP(p) TIMESTAMP WITHOUT TIME ZONE TIMESTAMP(p) WITHOUT TIME ZONE ``` **Example** ```sql CREATE TABLE test_sql_timestamp (val TIMESTAMP(3)); INSERT INTO test_sql_timestamp VALUES (CAST('2023-04-06 10:59:32.628' AS TIMESTAMP(3))); -- returns 2023-04-06 10:59:32.628 SELECT * FROM test_sql_timestamp; ``` Declare this type by using `TIMESTAMP(p)`, 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*. A space separates the date and time parts. 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 timezone, but this type is time-zone free. For more `java.time.Instant`-like semantics use `TIMESTAMP_LTZ`. `TIMESTAMP` acts like a pure string and isn’t related to a time zone of any kind, including UTC. `TIMESTAMP WITHOUT TIME ZONE` is a synonym for this type. ### Java **Syntax** ```java Timestamp DataTypes.TIMESTAMP(p) ``` **Examples** ```java import static org.apache.flink.table.api.DataTypes.*; DataType timestamp = TIMESTAMP(); DataType timestampWithPrecision = TIMESTAMP(p); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------------------------------|---------|----------|-------------------------| | java.time.LocalDateTime | ✓ | ✓ | Default | | java.sql.Timestamp | ✓ | ✓ | | | org.apache.flink.table.data.TimestampData | ✓ | ✓ | Internal data structure | ### Python **Syntax** ```python from pyflink.table.types import DataTypes DataTypes.TIMESTAMP(p) ``` **Examples** ```python from pyflink.table.types import DataTypes timestamp = DataTypes.TIMESTAMP() timestamp_with_precision = DataTypes.TIMESTAMP(p) ``` The precision specified in `p` must be `3`. **Formats** The following table shows examples of the TIMESTAMP type in different formats. | JSON for data type | ```json
{"type":"TIMESTAMP_WITHOUT_TIME_ZONE","nullable":true,"precision":3}
``` | |---------------------------|--------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
TIMESTAMP(3)
``` | | JSON for payload | ```none
"2023-04-06 10:59:32.628"
``` | | CLI/UI format for payload | ```none
2023-04-06 10:59:32.628
``` | | Statement literal | ```sql
TIMESTAMP '2023-04-06 10:59:32.628'
``` | ### TIMESTAMP_LTZ Represents a timestamp with the *local* timezone consisting of `year-month-day hour:minute:second[.fractional] zone` with up to nanosecond precision and values ranging from `0000-01-01 00:00:00.000000000 +14:59` to `9999-12-31 23:59:59.999999999 -14:59`. ### SQL **Syntax** ```sql TIMESTAMP_LTZ TIMESTAMP_LTZ(p) TIMESTAMP WITH LOCAL TIME ZONE TIMESTAMP(p) WITH LOCAL TIME ZONE ``` **Example** ```sql CREATE TABLE test_sql_timestamp_ltz (val TIMESTAMP_LTZ(3)); INSERT INTO test_sql_timestamp_ltz VALUES (CAST('2023-04-06 10:59:32.628' AS TIMESTAMP_LTZ(3))); -- returns 2023-04-06 10:59:32.628 SELECT * FROM test_sql_timestamp_ltz; ``` Declare this type by using `TIMESTAMP_LTZ(p)`, 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 timezone offset information is *not* stored physically in every datum. Instead, the type assumes `java.time.Instant` semantics in the UTC timezone at the edges of the table ecosystem. Every datum is interpreted in the local timezone configured in the current session for computation and visualization. This type fills the gap between time-zone free and time-zone mandatory timestamp types by allowing the interpretation of UTC timestamps according to the configured session timezone. `TIMESTAMP_LTZ` resembles a `TIMESTAMP` without a timezone, but the string always considers the sessions/query’s timezone. Internally, it is always in the UTC time zone. If you require the short format, prefer `TIMESTAMP_LTZ(3)`. `TIMESTAMP WITH LOCAL TIME ZONE` is a synonym for this type. ### Java **Syntax** ```java DataTypes.TIMESTAMP_LTZ(p) DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(p) ``` **Examples** ```java import static org.apache.flink.table.api.DataTypes.*; DataType timestamp_ltz = TIMESTAMP_LTZ(p); DataType timestamp_with_local_time_zone = TIMESTAMP_WITH_LOCAL_TIME_ZONE(p); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------------------------------|---------|----------|---------------------------------------------------------------------------------------------| | java.time.Instant | ✓ | ✓ | Default | | java.lang.Integer | ✓ | ✓ | Describes the number of seconds since Unix epoch. | | int | ✓ | (✓) | Describes the number of seconds since Unix epoch. Output only if type is not nullable. | | java.lang.Long | ✓ | ✓ | Describes the number of milliseconds since Unix epoch. | | long | ✓ | (✓) | Describes the number of milliseconds since Unix epoch. Output only if type is not nullable. | | java.sql.Timestamp | ✓ | ✓ | Describes the number of milliseconds since Unix epoch. | | org.apache.flink.table.data.TimestampData | ✓ | ✓ | Internal data structure | ### Python **Syntax** ```python from pyflink.table.types import DataTypes DataTypes.TIMESTAMP_LTZ(p) DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(p) ``` **Examples** ```python from pyflink.table.types import DataTypes timestamp_ltz = DataTypes.TIMESTAMP_LTZ(p) timestamp_with_local_time_zone = DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(p) ``` The precision specified in `p` must be `3`. **Formats** The following table shows examples of the TIMESTAMP_LTZ type in different formats. | JSON for data type | ```json
{"type":"TIMESTAMP_WITH_LOCAL_TIME_ZONE","nullable":true,"precision":3}
``` | |---------------------------|-----------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
TIMESTAMP(3) WITH LOCAL TIME ZONE
``` | | JSON for payload | ```none
"2023-04-06 11:06:47.224"
``` | | CLI/UI format for payload | ```none
2023-04-06 11:06:47.224
``` | | Statement literal | ```sql
CAST('2023-04-06 10:59:32.628' AS TIMESTAMP_LTZ(3))
``` | ### TIMESTAMP and TIMESTAMP_LTZ comparison Although TIMESTAMP and TIMESTAMP_LTZ are similarly named, they represent different concepts. TIMESTAMP_LTZ : - TIMESTAMP_LTZ in SQL is similar to the `Instant` class in Java. - TIMESTAMP_LTZ represents a *moment*, or a specific point in the UTC timeline. - TIMESTAMP_LTZ stores time as a UTC integer, which can be converted dynamically to every other timezone. - When printing or casting TIMESTAMP_LTZ as a character string, the `sql.local-time-zone` setting is considered. TIMESTAMP : - TIMESTAMP in SQL is similar to `LocalDateTime` in Java. - TIMESTAMP has no time zone or offset from UTC, so it can’t represent a moment. - TIMESTAMP stores time as character string, not related to any timezone. ### TIMESTAMP WITH TIME ZONE Represents a timestamp with time zone consisting of `year-month-day hour:minute:second[.fractional]` zone with up to nanosecond precision and values ranging from `0000-01-01 00:00:00.000000000 +14:59` to `9999-12-31 23:59:59.999999999 -14:59`. #### NOTE You can’t declare or populate `TIMESTAMP WITH TIME ZONE` in Flink SQL statements. It appears only as schema metadata. For example, it can appear when Flink infers a table’s schema from an external system or connector format, such as an Avro column that uses a timestamp-with-time-zone logical type. To store a timestamp together with an explicit zone offset that you can create through SQL, use [TIMESTAMP_LTZ](#flink-sql-timestamp-ltz) instead. ### SQL **Syntax** ```sql TIMESTAMP WITH TIME ZONE TIMESTAMP(p) WITH TIME ZONE ``` This is the standard SQL syntax for the type, but the Flink SQL parser doesn’t accept it in a `CREATE TABLE` statement or a `CAST` expression. Compared to TIMESTAMP_LTZ, the time zone offset information is stored physically in every datum. It is used individually for every computation, visualization, or communication to external systems. ### Java **Syntax** ```java OffsetDateTime DataTypes.TIMESTAMP_WITH_TIME_ZONE() DataTypes.TIMESTAMP_WITH_TIME_ZONE(p) ``` **Examples** ```java import static org.apache.flink.table.api.DataTypes.*; OffsetDateTime offsetDateTime = OffsetDateTime.of(2023, 4, 6, 10, 59, 32, 628, ZoneOffset.of("+08:00")); DataType timestamp_with_time_zone = TIMESTAMP_WITH_TIME_ZONE(); DataType timestamp_with_time_zone_with_precision = TIMESTAMP_WITH_TIME_ZONE(p); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |--------------------------|---------|----------|---------------------| | java.time.OffsetDateTime | ✓ | ✓ | Default | | java.time.ZonedDateTime | ✓ | | Ignores the zone ID | ### Python **Syntax** ```python from pyflink.table.types import DataTypes DataTypes.TIMESTAMP_WITH_TIME_ZONE(p) ``` **Examples** ```python from pyflink.table.types import DataTypes timestamp_with_time_zone = DataTypes.TIMESTAMP_WITH_TIME_ZONE(p) ``` Not supported. **Formats** The following table shows examples of the TIMESTAMP WITH TIME ZONE type in different formats. Because you can’t populate this type through Flink SQL, it has no payload format. | JSON for data type | ```json
{"type":"TIMESTAMP_WITH_TIME_ZONE","nullable":true,"precision":3}
``` | |---------------------------|-----------------------------------------------------------------------------------------| | CLI/UI format | ```sql
TIMESTAMP(3) WITH TIME ZONE
``` | | JSON for payload | Not applicable | | CLI/UI format for payload | Not applicable | | Statement literal | Not applicable | ## Collection data types ### ARRAY Represents an array of elements with same subtype. ### SQL **Syntax** ```sql ARRAY t ARRAY ``` **Example** ```sql CREATE TABLE test_sql_array (val ARRAY); INSERT INTO test_sql_array VALUES (ARRAY[1, 2, 3]); -- returns [1, 2, 3] SELECT * FROM test_sql_array; ``` Declare this type by using `ARRAY`, where `t` is the data type of the contained elements. Compared to the SQL standard, the maximum cardinality of an array cannot be specified and is fixed at *2,147,483,647*. Also, any valid type is supported as a subtype. `t ARRAY` is a synonym for being closer to the SQL standard. For example, `INT ARRAY` is equivalent to `ARRAY`. ### Java **Syntax** ```java t[] DataTypes.ARRAY(t) ``` **Examples** ```java int[] intArray = new int[] {1, 2, 3}; import static org.apache.flink.table.api.DataTypes.*; DataType array = ARRAY(t); DataType arrayWithPrecision = ARRAY(p); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |---------------------------------------|---------|----------|----------------------------------| | t[] | (✓) | (✓) | Default. Depends on the subtype. | | java.util.List | ✓ | ✓ | | | subclass of java.util.List | ✓ | | | | org.apache.flink.table.data.ArrayData | ✓ | ✓ | Internal data structure | ### Python **Syntax** ```python from pyflink.table.types import DataTypes DataTypes.ARRAY(t) ``` **Examples** ```python from pyflink.table.types import DataTypes array = DataTypes.ARRAY(t) array_with_precision = DataTypes.ARRAY(p) ``` **Formats** The following table shows examples of the ARRAY type in different formats. | JSON for data type | ```json
{"type":"ARRAY","nullable":true,"elementType":{"type":"INTEGER","nullable":true}}
``` | |---------------------------|---------------------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
ARRAY
``` | | JSON for payload | ```none
["1", "2", "3", null]
``` | | CLI/UI format for payload | ```none
[1, 2, 3, NULL]
``` | | Statement literal | ```sql
ARRAY[1, 2, 3]
``` | ### MAP Represents an associative array that maps keys (including `NULL`) to values (including `NULL`). ### SQL **Syntax** ```sql MAP ``` **Example** ```sql CREATE TABLE test_sql_map (val MAP); INSERT INTO test_sql_map VALUES (MAP['a', 1, 'b', 2, 'c', 3]); -- returns {a=1, b=2, c=3} SELECT * FROM test_sql_map; ``` Declare this type by using `MAP` where `kt` is the data type of the key elements and `vt` is the data type of the value elements. A map can’t contain duplicate keys. Each key can map to at most one value. There is no restriction of element types. It is the responsibility of the user to ensure uniqueness. The map type is an extension to the SQL standard. ### Java **Syntax** ```java Map DataTypes.MAP(kt, vt) ``` **Examples** ```java Map stringIntegerMap = new HashMap<>(); import static org.apache.flink.table.api.DataTypes.*; DataType map = MAP(kt, vt); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------------------------|---------|----------|-------------------------| | java.util.Map | ✓ | ✓ | Default | | subclass of java.util.Map | ✓ | | | | org.apache.flink.table.data.MapData | ✓ | ✓ | Internal data structure | ### Python **Syntax** ```python DataTypes.MAP(kt, vt) ``` **Examples** ```python from pyflink.table.types import DataTypes map = DataTypes.MAP(kt, vt) ``` **Formats** The following table shows examples of the MAP type in different formats. | JSON for data type | ```json
{"type":"MAP","nullable":true,"keyType":{"type":"INTEGER","nullable":true},"valueType":{"type":"VARCHAR","nullable":true,"length":2147483647}}
``` | |---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
MAP
``` | | JSON for payload | ```none
[["1", "a"], ["2", "b"], [null, "c"]]
``` | | CLI/UI format for payload | ```none
{1=a, 2=b, NULL=c}
``` | | Statement literal | ```sql
MAP[1, 'a', 2, 'b', 3, 'c']
``` | ### MULTISET Represents a multiset (=bag). ### SQL **Syntax** ```sql MULTISET t MULTISET ``` **Example** ```sql CREATE TABLE test_sql_multiset (val MULTISET); INSERT INTO test_sql_multiset SELECT COLLECT(val) FROM (VALUES 1, 2, 3, 2) AS t(val); -- returns {1, 2, 3, 2} SELECT * FROM test_sql_multiset; ``` Declare this type by using `MULTISET` where `t` is the data type of the contained elements. Unlike a set, the multiset allows for multiple instances for each of its elements with a common subtype. Each unique value (including `NULL`) is mapped to some multiplicity. There is no restriction of element types; it is the responsibility of the user to ensure uniqueness. `t MULTISET` is a synonym for being closer to the SQL standard. For example, `INT MULTISET` is equivalent to `MULTISET`. Unlike `ARRAY` and `MAP`, `MULTISET` doesn’t have a literal constructor syntax, such as `MULTISET[1, 2, 3]`. Construct a `MULTISET` value by aggregating rows with the `COLLECT` function, as shown in the preceding example. ### Java **Syntax** ```java DataTypes.MULTISET(t) ``` **Examples** ```java import static org.apache.flink.table.api.DataTypes.*; DataType multiset = MULTISET(t); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------------------------------------|---------|----------|---------------------------------------------------------| | java.util.Map | ✓ | ✓ | Default. Assigns each value to an integer multiplicity. | | subclass of java.util.Map | ✓ | | | | org.apache.flink.table.data.MapData | ✓ | ✓ | Internal data structure | ### Python **Syntax** ```python DataTypes.MULTISET(t) ``` **Examples** ```python from pyflink.table.types import DataTypes multiset = DataTypes.MULTISET(t) ``` **Formats** The following table shows examples of the MULTISET type in different formats. | JSON for data type | ```json
{"type":"MULTISET","nullable":true,"elementType":{"type":"INTEGER","nullable":true}}
``` | |---------------------------|------------------------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
MULTISET
``` | | JSON for payload | ```none
[["a", "1"], ["b", "2"], [null, "1"]]
``` | | CLI/UI format for payload | ```none
{a=1, b=2, NULL=1}
``` | | Statement literal | No literal syntax available | ### ROW Represents a sequence of fields. ### SQL **Syntax** ```sql ROW ROW ROW(name0 type0, name1 type1, ...) ROW(name0 type0 'description0', name1 type1 'description1', ...) ``` **Example** ```sql CREATE TABLE test_sql_row (val ROW); INSERT INTO test_sql_row VALUES (('Alice', 30)), (('Bob',45)); -- returns ('Alice', 30), ('Bob', 45) SELECT * FROM test_sql_row; ``` Declare this type by using `ROW`, where `n` is the unique name of a field, `t` is the logical type of a field, `d` is the description of a field. A field consists of a field name, field type, and an optional description. The most specific type of a row of a table is a row type. In this case, each column of the row corresponds to the field of the row type that has the same ordinal position as the column. To create a table with a row type, use the following syntax: ```sql CREATE TABLE table_with_row_types ( `Customer` ROW, `Order` ROW ); ``` To insert a row into a table with a row type, use the following syntax: ```sql INSERT INTO table_with_row_types VALUES (('Alice', 30), (101, 'Book')), (('Bob', 25), (102, 'Laptop')), (('Charlie', 35), (103, 'Phone')), (('Diana', 28), (104, 'Tablet')), (('Eve', 22), (105, 'Headphones')); ``` To work with fields from a row, use dot notation: ```sql SELECT `Customer`.name, `Customer`.age, `Order`.id, `Order`.title FROM table_with_row_types WHERE `Customer`.age > 30; ``` Compared to the SQL standard, an optional field description simplifies the handling with complex structures. A row type is similar to the `STRUCT` type known from other non-standard-compliant frameworks. `ROW(...)` is a synonym for being closer to the SQL standard. For example, `ROW(fieldOne INT, fieldTwo BOOLEAN)` is equivalent to `ROW`. The preceding syntax declares a `ROW` *type*. To construct a `ROW` *value*, list the field values, without field names or types, by using `ROW(v0, v1, ...)` or the equivalent shorthand `(v0, v1, ...)`: ```sql SELECT ROW('Alice', 30); SELECT ('Alice', 30); ``` A `ROW` value literal can’t assign names to its fields. The field names come from the target `ROW` type, for example, the column type in a `CREATE TABLE` statement or an explicit `CAST`. If the fields of the data type contain characters other than `[A-Za-z_]`, use escaping notation. Double backticks escape the backtick character, for example: ```sql ROW<`a-b` INT, b STRING, `weird_col``_umn` STRING> ``` Rows fields can contain comments, for example: ```sql {"type":"ROW","nullable":true,"fields":[{"name":"a","fieldType":{"type":"INTEGER","nullable":true},"description":"hello"}]} ``` Format using single quotes. Double single quotes escape single quotes, for example: ```sql ROW ``` ### Java **Syntax** ```java DataTypes.ROW(DataTypes.FIELD(n0, t0), DataTypes.FIELD(n1, t1), ...) DataTypes.ROW(DataTypes.FIELD(n0, t0, d0), DataTypes.FIELD(n1, t1, d1), ...) ``` **Examples** ```java import static org.apache.flink.table.api.DataTypes.*; DataType row = ROW(DataTypes.FIELD("name", DataTypes.STRING()), DataTypes.FIELD("age", DataTypes.INT())); DataType rowWithDescription = ROW(DataTypes.FIELD("name", DataTypes.STRING(), "name"), DataTypes.FIELD("age", DataTypes.INT(), "age")); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------------------------|---------|----------|-------------------------| | org.apache.flink.types.Row | ✓ | ✓ | Default | | org.apache.flink.table.data.RowData | ✓ | ✓ | Internal data structure | ### Python **Syntax** ```python from pyflink.table import DataTypes DataTypes.ROW([DataTypes.FIELD(..), DataTypes.FIELD(..), ...]) ``` **Examples** ```python from pyflink.table import DataTypes row = DataTypes.ROW([ DataTypes.FIELD("id", DataTypes.INT()), DataTypes.FIELD("data", DataTypes.STRING()) ]) ``` **Formats** The following table shows examples of the ROW type in different formats. | JSON for data type | ```json
{"type":"ROW","nullable":true,"fields":[{"name":"a","fieldType":{"type":"INTEGER","nullable":true}},{"name":"b","fieldType":{"type":"VARCHAR","nullable":true,"length":2147483647}}]}
``` | |---------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | CLI/UI format | ```sql
ROW
``` | | JSON for payload | ```none
{"a": "1", "b": "value"}
``` | | CLI/UI format for payload | ```none
(1, value)
``` | | Statement literal | ```sql
ROW(1, 'value')
``` | ## Other data types ### BOOLEAN Represents a boolean with a (possibly) three-valued logic of `TRUE`, `FALSE`, and `UNKNOWN`. ### SQL **Syntax** ```sql BOOLEAN ``` **Example** ```sql CREATE TABLE test_sql_boolean (val BOOLEAN); INSERT INTO test_sql_boolean VALUES (TRUE), (FALSE), (UNKNOWN); -- returns TRUE, FALSE, NULL SELECT * FROM test_sql_boolean; ``` ### Java **Syntax** ```java boolean DataTypes.BOOLEAN() ``` **Examples** ```java boolean booleanValue = true; import static org.apache.flink.table.api.DataTypes.*; DataType boolean = BOOLEAN(); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |-------------------|---------|----------|--------------------------------------| | java.lang.Boolean | ✓ | ✓ | Default | | boolean | ✓ | (✓) | Output only if type is not nullable. | ### Python **Syntax** ```python DataTypes.BOOLEAN() ``` **Examples** ```python from pyflink.table import DataTypes boolean = DataTypes.BOOLEAN() boolean_value = True ``` **Formats** The following table shows examples of the BOOLEAN type in different formats. | JSON for data type | ```json
{"type":"BOOLEAN","nullable":true}
``` | |---------------------------|----------------------------------------------------------| | CLI/UI format | ```sql
BOOLEAN
``` | | JSON for payload | ```none
true
``` | | CLI/UI format for payload | ```none
TRUE
``` | | Statement literal | ```sql
TRUE
``` | ### NULL Data type for representing untyped `NULL` values. ### SQL **Syntax** ```sql NULL ``` The NULL type is an extension to the SQL standard. A NULL type has no other value except `NULL`, so it can be cast to any nullable type similar to JVM semantics. This type helps in representing unknown types in API calls that use a `NULL` literal, and for bridging to formats such as JSON or Avro that also define this type. This type is rarely useful in practice and is described here only for completeness. ### Java **Syntax** ```java DataTypes.NULL() ``` **Examples** ```java import static org.apache.flink.table.api.DataTypes.*; DataType null = NULL(); ``` **Bridging to JVM types** | Java Type | Input | Output | Notes | |------------------|---------|----------|-------------------------| | java.lang.Object | ✓ | ✓ | Default | | any class | | (✓) | Any non-primitive type. | ### Python **Syntax** ```python DataTypes.NULL() ``` **Examples** ```python from pyflink.table import DataTypes null = DataTypes.NULL() ``` **Formats** The following table shows examples of the NULL type in different formats. | JSON for data type | ```json
{"type":"NULL"}
``` | |---------------------------|---------------------------------------| | CLI/UI format | ```sql
NULL
``` | | JSON for payload | ```none
null
``` | | CLI/UI format for payload | ```none
NULL
``` | | Statement literal | ```sql
CAST(NULL AS INT)
``` | ## Casting Flink SQL can perform casting between a defined input type and target type. While some casting operations can always succeed regardless of the input value, others can fail at runtime when there’s no way to create a value for the target type. For example, it’s always possible to convert `INT` to `STRING`, but you can’t always convert a `STRING` to `INT`. During the planning stage, the query validator rejects queries for invalid type pairs with a `ValidationException`, for example, when trying to cast a `TIMESTAMP` to an `INTERVAL`. Valid type pairs that can fail at runtime are accepted by the query validator, but this requires you to handle cast failures correctly. In Flink SQL, casting can be performed by using one of these two built-in functions: - [CAST](functions/comparison-functions.md#flink-sql-cast-function): The regular cast function defined by the SQL standard. It can fail the job if the cast operation is fallible and the provided input is not valid. Type inference preserves the nullability of the input type. - [TRY_CAST](functions/comparison-functions.md#flink-sql-try-cast-function): An extension to the regular cast function that returns `NULL` if the cast operation fails. Its return type is always nullable. For example: ```sql -- returns 42 of type INT NOT NULL SELECT CAST('42' AS INT); -- returns NULL of type VARCHAR SELECT CAST(NULL AS VARCHAR); -- throws an exception and fails the job SELECT CAST('non-number' AS INT); -- returns 42 of type INT SELECT TRY_CAST('42' AS INT); -- returns NULL of type VARCHAR SELECT TRY_CAST(NULL AS VARCHAR); -- returns NULL of type INT SELECT TRY_CAST('non-number' AS INT); -- returns 0 of type INT NOT NULL SELECT COALESCE(TRY_CAST('non-number' AS INT), 0); ``` The following matrix shows the supported cast pairs, where “Y” means supported, “!” means fallible, and “N” means unsupported: | Input / Target | CHAR¹ / VARCHAR¹ / STRING | BINARY¹ / VARBINARY¹ / BYTES | BOOLEAN | DECIMAL | TINYINT | SMALLINT | INTEGER | BIGINT | FLOAT | DOUBLE | DATE | TIME | TIMESTAMP | TIMESTAMP_LTZ | INTERVAL | ARRAY | MULTISET | MAP | ROW | |----------------------------|-----------------------------|--------------------------------|-----------|-----------|-----------|------------|-----------|----------|---------|----------|--------|--------|-------------|-----------------|------------|---------|------------|-------|-------| | CHAR / VARCHAR / STRING | Y | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | N | N | N | N | | BINARY / VARBINARY / BYTES | Y | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | | BOOLEAN | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | | DECIMAL | Y | N | N | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | | TINYINT | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | | SMALLINT | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | N | N | N | N | N | | INTEGER | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁵ | N | N | N | N | | BIGINT | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N² | N² | Y⁶ | N | N | N | N | | FLOAT | Y | N | N | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | | DOUBLE | Y | N | N | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | | DATE | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | | TIME | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | | TIMESTAMP | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | | TIMESTAMP_LTZ | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | | INTERVAL | Y | N | N | N | N | N | Y⁵ | Y⁶ | N | N | N | N | N | N | Y | N | N | N | N | | ARRAY | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | | MULTISET | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | | MAP | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | | ROW | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | Notes: 1. All the casting to constant length or variable length also trims and pads, according to the type definition. 2. `TO_TIMESTAMP` and `TO_TIMESTAMP_LTZ` must be used instead of `CAST`/ `TRY_CAST`. 3. Supported iff the children type pairs are supported. Fallible iff the children type pairs are fallible. 4. Supported iff the `RAW` class and serializer are equals. 5. Supported iff `INTERVAL` is a `MONTH TO YEAR` range. 6. Supported iff `INTERVAL` is a `DAY TO TIME` range. #### NOTE A cast of a `NULL` value always returns `NULL`, regardless of whether the function used is [CAST](functions/comparison-functions.md#flink-sql-cast-function) or [TRY_CAST](functions/comparison-functions.md#flink-sql-try-cast-function). ## Data type extraction In many locations in the API, Flink tries to extract data types automatically from class information by using reflection to avoid repetitive manual schema work. But extracting a data type using reflection is not always successful, because logical information might be missing. In these cases, it can be necessary to add additional information close to a class or field declaration for supporting the extraction logic. The following table lists classes that map implicitly to a data type without requiring further information. Other JVM bridging classes require the [@DataTypeHint](../concepts/user-defined-functions.md#flink-sql-udfs-type-inference-data-type-hints) annotation. | Class | Data Type | |--------------------------|-----------------------------| | boolean | BOOLEAN NOT NULL | | byte | TINYINT NOT NULL | | byte[] | BYTES | | double | DOUBLE NOT NULL | | float | FLOAT NOT NULL | | int | INT NOT NULL | | java.lang.Boolean | BOOLEAN | | java.lang.Byte | TINYINT | | java.lang.Double | DOUBLE | | java.lang.Float | FLOAT | | java.lang.Integer | INT | | java.lang.Long | BIGINT | | java.lang.Short | SMALLINT | | java.lang.String | STRING | | java.sql.Date | DATE | | java.sql.Time | TIME(0) | | java.sql.Timestamp | TIMESTAMP(9) | | java.time.Duration | INTERVAL SECOND(9) | | java.time.Instant | TIMESTAMP_LTZ(9) | | java.time.LocalDate | DATE | | java.time.LocalTime | TIME(9) | | java.time.LocalDateTime | TIMESTAMP(9) | | java.time.OffsetDateTime | TIMESTAMP(9) WITH TIME ZONE | | java.time.Period | INTERVAL YEAR(4) TO MONTH | | java.util.Map | MAP | | short | SMALLINT NOT NULL | | structured type T | anonymous structured type T | | long | BIGINT NOT NULL | | T[] | ARRAY | ## Related content - [DDL Statements](../concepts/statements.md#flink-sql-statements) - [Flink SQL Queries](queries/overview.md#flink-sql-queries) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).