CREATE MATERIALIZED TABLE Statement in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables creating materialized tables that combine a table definition with a continuous query in a single persistent object. Use materialized tables for long-running streaming queries that act as incremental materialized views. For more information, see Materialized Tables.

Syntax

CREATE MATERIALIZED TABLE [IF NOT EXISTS] [catalog_name.][db_name.]table_name
  [(
    { <physical_column_definition> |
      <metadata_column_definition> |
      <computed_column_definition> }[ , ...n]
    [ <watermark_definition> ]
    [ <table_constraint> ][ , ...n]
  )]
  [COMMENT table_comment]
  [DISTRIBUTED BY (column_name1, column_name2, ...) INTO n BUCKETS]
  [WITH (key1=value1, key2=value2, ...)]
  [START_MODE = <start_mode_value>]
  AS <select_query>

Description

The CREATE MATERIALIZED TABLE statement creates a new persistent materialized table in the current or specified catalog. When a materialized table is created, Flink performs the following steps:

Creates a backing Apache Kafka® topic for storing query results.
Registers the output schema in Schema Registry.
Starts a continuous query that processes data from the source tables and writes results to the backing topic.

Unlike a regular CREATE TABLE combined with an INSERT INTO statement, a materialized table is a single declarative object that owns both the table definition and the continuous query. This makes it possible to evolve the pipeline in place using CREATE OR ALTER MATERIALIZED TABLE.

If a materialized table with the same name already exists, the statement fails unless you specify IF NOT EXISTS.

Note

The CREATE MATERIALIZED TABLE statement can only create new tables. To convert an existing table, use CREATE OR ALTER MATERIALIZED TABLE.

Explicit and inferred schemas

You can define the schema explicitly by listing column definitions, or let Flink infer the schema from the SELECT query.

Explicit schema: Declare the columns, types and any constraints yourself.
Inferred schema: Omit the column list and let Flink derive the schema from the query output.

Declaring the schema explicitly is recommended. The types a query infers are hard to predict. Nullability, decimal precision and scale, plus string length are all derived automatically from the query and are not visible in the SQL. Those choices form a data contract that affects how the materialized table can evolve later.

A strict inferred type, such as NOT NULL or a tight precision or length, locks every future version of the query into producing the same type. A later change that could yield a null, a longer string or a higher-precision number is then rejected.
Looser types, such as a nullable column, STRING or a generous decimal precision, keep the table evolvable but give downstream consumers weaker guarantees.

Declaring the schema makes this a deliberate, per-column choice rather than one that is inferred for you.

This guidance is for creating a new materialized table. When you convert an existing table with CREATE OR ALTER MATERIALIZED TABLE, match the schema already registered for the topic. Choosing fresh types risks changing the schema during conversion.

START_MODE

The optional START_MODE clause controls how much historical data is processed when the materialized table is first created. If omitted, the default is RESUME_OR_FROM_BEGINNING, which processes all available historical data on initial creation.

For the full list of START_MODE values and detailed behavior, see START_MODE.

Usage

Create with explicit schema

CREATE MATERIALIZED TABLE high_value_orders (
  `order_id` STRING,
  `customer_id` INT,
  `price` DOUBLE
) AS
SELECT order_id, customer_id, price
FROM examples.marketplace.orders
WHERE price > 50.00;

Create with inferred schema

Inferred schema is shown for completeness. For production tables, declaring the schema explicitly is recommended, so the types are a deliberate choice. For more information, see Explicit and inferred schemas.

CREATE MATERIALIZED TABLE all_orders
AS
SELECT * FROM examples.marketplace.orders;

Create using WITH properties

CREATE MATERIALIZED TABLE orders_avro (
  `order_id` STRING,
  `customer_id` INT,
  `price` DOUBLE
)
WITH (
  'value.format' = 'avro-registry'
)
AS
SELECT order_id, customer_id, price
FROM examples.marketplace.orders;

Create with a JOIN

CREATE MATERIALIZED TABLE customer_orders (
  `order_id` STRING,
  `customer_name` STRING,
  `price` DOUBLE
) AS
SELECT o.order_id, c.name AS customer_name, o.price
FROM examples.marketplace.orders o
JOIN examples.marketplace.customers c ON o.customer_id = c.customer_id;

WITH options

Materialized tables support the same WITH options as CREATE TABLE for configuring the backing Kafka topic. Both the key and value of the expression key1=val1 are string literals.

You can change an existing materialized table’s property values by using ALTER MATERIALIZED TABLE.

changelog.mode	connector	error-handling.log.target
error-handling.mode	kafka.cleanup-policy	kafka.compaction.time
kafka.max-message-size	kafka.retention.size	kafka.retention.time
key.fields-prefix	key.format	key.format.schema-context
scan.bounded.mode	scan.bounded.timestamp-millis	scan.startup.mode
value.fields-include	value.format	value.format.schema-context

Limitations

For the full list of current limitations, see Materialized Tables limitations.