<a id="flink-sql-create-materialized-table"></a>

# 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](../../concepts/materialized-tables.md#flink-sql-materialized-tables).

## Syntax

```none
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](create-or-alter-materialized-table.md#flink-sql-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](create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table).

<a id="flink-sql-mt-explicit-schema"></a>

### 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](create-or-alter-materialized-table.md#flink-sql-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](create-or-alter-materialized-table.md#flink-sql-start-mode).

## Usage

### Create with explicit schema

```sql
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](#flink-sql-mt-explicit-schema).

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

### Create using WITH properties

```sql
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

```sql
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](create-table.md#flink-sql-with-options) 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](alter-materialized-table.md#flink-sql-alter-materialized-table).

| [changelog.mode](create-table.md#flink-sql-create-table-with-changelog-mode)                 | [connector](create-table.md#flink-sql-create-table-with-connector)                                         | [error-handling.log.target](create-table.md#flink-sql-create-table-with-error-handling-log-target)     |
|----------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------|
| [error-handling.mode](create-table.md#flink-sql-create-table-with-error-handling-mode)       | [kafka.cleanup-policy](create-table.md#flink-sql-create-table-with-kafka-cleanup-policy)                   | [kafka.compaction.time](create-table.md#flink-sql-create-table-with-kafka-compaction-time)             |
| [kafka.max-message-size](create-table.md#flink-sql-create-table-with-kafka-max-message-size) | [kafka.retention.size](create-table.md#flink-sql-create-table-with-kafka-retention-size)                   | [kafka.retention.time](create-table.md#flink-sql-create-table-with-kafka-retention-time)               |
| [key.fields-prefix](create-table.md#flink-sql-create-table-with-key-fields-prefix)           | [key.format](create-table.md#flink-sql-create-table-with-key-format)                                       | [key.format.schema-context](create-table.md#flink-sql-create-table-with-key-format-schema-context)     |
| [scan.bounded.mode](create-table.md#flink-sql-create-table-with-scan-bounded-mode)           | [scan.bounded.timestamp-millis](create-table.md#flink-sql-create-table-with-scan-bounded-timestamp-millis) | [scan.startup.mode](create-table.md#flink-sql-create-table-with-scan-startup-mode)                     |
| [value.fields-include](create-table.md#flink-sql-create-table-with-value-fields-include)     | [value.format](create-table.md#flink-sql-create-table-with-value-format)                                   | [value.format.schema-context](create-table.md#flink-sql-create-table-with-value-format-schema-context) |

## Limitations

For the full list of current limitations, see
[Materialized Tables limitations](../../concepts/materialized-tables.md#flink-sql-materialized-tables-limitations).

## Related content

- [CREATE OR ALTER MATERIALIZED TABLE](create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table)
- [ALTER MATERIALIZED TABLE](alter-materialized-table.md#flink-sql-alter-materialized-table)
- [DROP MATERIALIZED TABLE](drop-materialized-table.md#flink-sql-drop-materialized-table)
- [Materialized Tables concept](../../concepts/materialized-tables.md#flink-sql-materialized-tables)
- [SHOW CREATE MATERIALIZED TABLE](show.md#flink-sql-show-create-materialized-table)
- [Manage |af-long| with Terraform](../../operate-and-deploy/terraform.md#flink-sql-terraform)

#### 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).