CREATE statements are used to register a table/view/function into current or specified Catalog. A registered table/view/function can be used in SQL queries.
Flink SQL supports the following CREATE statements for now:
CREATE statements can be executed with the executeSql()
method of the TableEnvironment
. The executeSql()
method returns ‘OK’ for a successful CREATE operation, otherwise will throw an exception.
The following examples show how to run a CREATE statement in TableEnvironment
.
CREATE statements can be executed with the execute_sql()
method of the TableEnvironment
. The execute_sql()
method returns ‘OK’ for a successful CREATE operation, otherwise will throw an exception.
The following examples show how to run a CREATE statement in TableEnvironment
.
CREATE statements can be executed in SQL CLI.
The following examples show how to run a CREATE statement in SQL CLI.
The following grammar gives an overview about the available syntax:
The statement above creates a table with the given name. If a table with the same name already exists in the catalog, an exception is thrown.
Physical / Regular Columns
Physical columns are regular columns known from databases. They define the names, the types, and the order of fields in the physical data. Thus, physical columns represent the payload that is read from and written to an external system. Connectors and formats use these columns (in the defined order) to configure themselves. Other kinds of columns can be declared between physical columns but will not influence the final physical schema.
The following statement creates a table with only regular columns:
Metadata Columns
Metadata columns are an extension to the SQL standard and allow to access connector and/or format specific
fields for every row of a table. A metadata column is indicated by the METADATA
keyword. For example,
a metadata column can be be used to read and write the timestamp from and to Kafka records for time-based
operations. The connector and format documentation lists the
available metadata fields for every component. However, declaring a metadata column in a table’s schema
is optional.
The following statement creates a table with an additional metadata column that references the metadata field timestamp
:
Every metadata field is identified by a string-based key and has a documented data type. For example,
the Kafka connector exposes a metadata field with key timestamp
and data type TIMESTAMP(3) WITH LOCAL TIME ZONE
that can be used for both reading and writing records.
In the example above, the metadata column record_time
becomes part of the table’s schema and can be
transformed and stored like a regular column:
For convenience, the FROM
clause can be omitted if the column name should be used as the identifying metadata key:
For convenience, the runtime will perform an explicit cast if the data type of the column differs from the data type of the metadata field. Of course, this requires that the two data types are compatible.
By default, the planner assumes that a metadata column can be used for both reading and writing. However,
in many cases an external system provides more read-only metadata fields than writable fields. Therefore,
it is possible to exclude metadata columns from persisting using the VIRTUAL
keyword.
In the example above, the offset
is a read-only metadata column and excluded from the query-to-sink
schema. Thus, source-to-query schema (for SELECT
) and query-to-sink (for INSERT INTO
) schema differ:
Computed Columns
Computed columns are virtual columns that are generated using the syntax column_name AS computed_column_expression
.
A computed column evaluates an expression that can reference other columns declared in the same table. Both physical columns and metadata columns can be accessed if they preceed the computed column in the schema declaration. The column itself is not physically stored within the table. The column’s data type is derived automatically from the given expression and does not have to be declared manually.
The planner will transform computed columns into a regular projection after the source. For optimization or watermark strategy push down, the evaluation might be spread across operators, performed multiple times, or skipped if not needed for the given query.
For example, a computed column could be defined as:
The expression may contain any combination of columns, constants, or functions. The expression cannot contain a subquery.
Computed columns are commonly used in Flink for defining time attributes
in CREATE TABLE
statements.
proc AS PROCTIME()
using the system’s PROCTIME()
function.WATERMARK
declaration. For example, the computed column can be used
if the original field is not TIMESTAMP(3)
type or is nested in a JSON string.Similar to virtual metadata columns, computed columns are excluded from persisting. Therefore, a computed
column cannot be the target of an INSERT INTO
statement. Thus, source-to-query schema (for SELECT
)
and query-to-sink (for INSERT INTO
) schema differ:
WATERMARK
The WATERMARK
clause defines the event time attributes of a table and takes the form WATERMARK FOR rowtime_column_name AS watermark_strategy_expression
.
The rowtime_column_name
defines an existing column that is marked as the event time attribute of the table. The column must be of type TIMESTAMP(3)
and be a top-level column in the schema. It may be a computed column.
The watermark_strategy_expression
defines the watermark generation strategy. It allows arbitrary non-query expression, including computed columns, to calculate the watermark. The expression return type must be TIMESTAMP(3), which represents the timestamp since the Epoch.
The returned watermark will be emitted only if it is non-null and its value is larger than the previously emitted local watermark (to preserve the contract of ascending watermarks). The watermark generation expression is evaluated by the framework for every record.
The framework will periodically emit the largest generated watermark. If the current watermark is still identical to the previous one, or is null, or the value of the returned watermark is smaller than that of the last emitted one, then no new watermark will be emitted.
Watermark is emitted in an interval defined by pipeline.auto-watermark-interval
configuration.
If watermark interval is 0ms
, the generated watermarks will be emitted per-record if it is not null and greater than the last emitted one.
When using event time semantics, tables must contain an event time attribute and watermarking strategy.
Flink provides several commonly used watermark strategies.
Strictly ascending timestamps: WATERMARK FOR rowtime_column AS rowtime_column
.
Emits a watermark of the maximum observed timestamp so far. Rows that have a timestamp bigger to the max timestamp are not late.
Ascending timestamps: WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '0.001' SECOND
.
Emits a watermark of the maximum observed timestamp so far minus 1. Rows that have a timestamp bigger or equal to the max timestamp are not late.
Bounded out of orderness timestamps: WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL 'string' timeUnit
.
Emits watermarks, which are the maximum observed timestamp minus the specified delay, e.g., WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '5' SECOND
is a 5 seconds delayed watermark strategy.
PRIMARY KEY
Primary key constraint is a hint for Flink to leverage for optimizations. It tells that a column or a set of columns of a table or a view are unique and they do not contain null. Neither of columns in a primary can be nullable. Primary key therefore uniquely identify a row in a table.
Primary key constraint can be either declared along with a column definition (a column constraint) or as a single line (a table constraint). For both cases, it should only be declared as a singleton. If you define multiple primary key constraints at the same time, an exception would be thrown.
Validity Check
SQL standard specifies that a constraint can either be ENFORCED
or NOT ENFORCED
. This controls if the constraint checks are performed on the incoming/outgoing data.
Flink does not own the data therefore the only mode we want to support is the NOT ENFORCED
mode.
It is up to the user to ensure that the query enforces key integrity.
Flink will assume correctness of the primary key by assuming that the columns nullability is aligned with the columns in primary key. Connectors should ensure those are aligned.
Notes: In a CREATE TABLE statement, creating a primary key constraint will alter the columns nullability, that means, a column with primary key constraint is not nullable.
PARTITIONED BY
Partition the created table by the specified columns. A directory is created for each partition if this table is used as a filesystem sink.
WITH
OptionsTable properties used to create a table source/sink. The properties are usually used to find and create the underlying connector.
The key and value of expression key1=val1
should both be string literal. See details in Connect to External Systems for all the supported table properties of different connectors.
Notes: The table name can be of three formats: 1. catalog_name.db_name.table_name
2. db_name.table_name
3. table_name
. For catalog_name.db_name.table_name
, the table would be registered into metastore with catalog named “catalog_name” and database named “db_name”; for db_name.table_name
, the table would be registered into the current catalog of the execution table environment and database named “db_name”; for table_name
, the table would be registered into the current catalog and database of the execution table environment.
Notes: The table registered with CREATE TABLE
statement can be used as both table source and table sink, we can not decide if it is used as a source or sink until it is referenced in the DMLs.
LIKE
The LIKE
clause is a variant/combination of SQL features (Feature T171, “LIKE clause in table definition” and Feature T173, “Extended LIKE clause in table definition”). The clause can be used to create a table based on a definition of an existing table. Additionally, users
can extend the original table or exclude certain parts of it. In contrast to the SQL standard the clause must be defined at the top-level of a CREATE statement. That is because the clause applies to multiple parts of the definition and not only to the schema part.
You can use the clause to reuse (and potentially overwrite) certain connector properties or add watermarks to tables defined externally. For example, you can add a watermark to a table defined in Apache Hive.
Consider the example statement below:
The resulting table Orders_with_watermark
will be equivalent to a table created with a following statement:
The merging logic of table features can be controlled with like options
.
You can control the merging behavior of:
with three different merging strategies:
Additionally, you can use the INCLUDING/EXCLUDING ALL
option to specify what should be the strategy if there was no specific strategy defined, i.e. if you use EXCLUDING ALL INCLUDING WATERMARKS
only the watermarks will be included from the source table.
Example:
If you provide no like options, INCLUDING ALL OVERWRITING OPTIONS
will be used as a default.
NOTE You cannot control the behavior of merging physical columns. Those will be merged as if you applied the INCLUDING
strategy.
NOTE The source_table
can be a compound identifier. Thus, it can be a table from a different catalog or database: e.g. my_catalog.my_db.MyTable
specifies table MyTable
from catalog MyCatalog
and database my_db
; my_db.MyTable
specifies table MyTable
from current catalog and database my_db
.
Create a catalog with the given catalog properties. If a catalog with the same name already exists, an exception is thrown.
WITH OPTIONS
Catalog properties used to store extra information related to this catalog.
The key and value of expression key1=val1
should both be string literal.
Check out more details at Catalogs.
Create a database with the given database properties. If a database with the same name already exists in the catalog, an exception is thrown.
IF NOT EXISTS
If the database already exists, nothing happens.
WITH OPTIONS
Database properties used to store extra information related to this database.
The key and value of expression key1=val1
should both be string literal.
Create a view with the given query expression. If a view with the same name already exists in the catalog, an exception is thrown.
TEMPORARY
Create temporary view that has catalog and database namespaces and overrides views.
IF NOT EXISTS
If the view already exists, nothing happens.
Create a catalog function that has catalog and database namespaces with the identifier and optional language tag. If a function with the same name already exists in the catalog, an exception is thrown.
If the language tag is JAVA/SCALA, the identifier is the full classpath of the UDF. For the implementation of Java/Scala UDF, please refer to User-defined Functions for more details.
If the language tag is PYTHON, the identifier is the fully qualified name of the UDF, e.g. pyflink.table.tests.test_udf.add
. For the implementation of Python UDF, please refer to Python UDFs for more details.
If the language tag is PYTHON, however the current program is written in Java/Scala or pure SQL, then you need to configure the Python dependencies.
TEMPORARY
Create temporary catalog function that has catalog and database namespaces and overrides catalog functions.
TEMPORARY SYSTEM
Create temporary system function that has no namespace and overrides built-in functions
IF NOT EXISTS
If the function already exists, nothing happens.
LANGUAGE JAVA|SCALA|PYTHON
Language tag to instruct Flink runtime how to execute the function. Currently only JAVA, SCALA and PYTHON are supported, the default language for a function is JAVA.