CREATE OR ALTER MATERIALIZED TABLE Statement in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables evolving a materialized table in place by using the CREATE OR ALTER MATERIALIZED TABLE statement. This statement creates a new materialized table if it does not exist, or triggers an evolution of an existing one.

An evolution updates the continuous query, schema, or both, while keeping the same output topic. This automates the complex manual process of stopping a statement, carrying over offsets, and migrating downstream consumers.

For details on table definition options like schemas, WITH properties, and DISTRIBUTED BY, see CREATE MATERIALIZED TABLE.

Syntax

CREATE OR ALTER MATERIALIZED TABLE [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

When you run CREATE OR ALTER MATERIALIZED TABLE:

If no table exists with that name, a new materialized table is created, the same behavior as CREATE MATERIALIZED TABLE.
If a materialized table already exists, an evolution is triggered.
If a regular table already exists, Flink adopts it as a materialized table in place. For details, see Adopt an existing table.

How evolution works

An evolution performs an in-place migration:

The existing continuous query is stopped.
A new continuous query is created with the updated query, schema, and configuration.
The new query begins processing data according to the START_MODE setting.
Results are written to the same output Kafka topic as before.

State handling

All existing Flink processing state, including aggregation counts, join state, and window state, is discarded when an evolution is triggered. A new state is built from scratch based on the reprocessing settings specified by START_MODE.

For stateful queries, like GROUP BY aggregations, the results are recalculated by reprocessing the source data, not by migrating the previous state.

Important behavior

Not idempotent: Running the same CREATE OR ALTER command always triggers a new evolution, even if nothing has changed. Use caution when running this command in automated scripts to avoid unintentional reprocessing.
Concurrent evolutions are rejected: If an evolution is already in progress for the same materialized table, a second CREATE OR ALTER command is rejected with an error.
No automatic rollback: If the new query fails at runtime, for example, due to a permission error or a UDF failure, the materialized table enters a FAILED state. The previous version is not restored automatically.

Adopt an existing table as a materialized table

You can turn a table that already exists into a materialized table, instead of creating a new topic. This helps when the destination topic was created outside Flink, for example by a platform team through Terraform or a CI/CD pipeline. In that case you want Flink to keep the topic up to date without recreating it. In Confluent Cloud every Kafka topic already appears in Flink as a table.

Declare the schema explicitly when you adopt a table. Matching the columns, types, primary key, and watermark the topic already has keeps the schema registered in Schema Registry unchanged, so the contract that consumers rely on does not change. If you let the query infer the schema instead, it can write a different schema to Schema Registry. For more on this choice, see Explicit and inferred schemas.

The recommended flow is to print the existing definition first, then declare the same schema in a CREATE OR ALTER MATERIALIZED TABLE statement with the query that keeps the table up to date:

SHOW CREATE TABLE user_spending;

CREATE OR ALTER MATERIALIZED TABLE user_spending (
  customer_id INT,
  total_spend DOUBLE
) AS
SELECT customer_id, SUM(price) AS total_spend
FROM examples.marketplace.orders
GROUP BY customer_id;

When you adopt a table, some things carry over and some are defined fresh:

The topic and its data stay. The materialized table reuses the same Kafka topic and keeps the data already in it. Any consumer reading that topic keeps working.
The schema must stay compatible with the existing topic. Flink checks the schema your statement produces against the schema already registered for the topic. If the schemas are not compatible, for example if a column type changes or a nullable column becomes NOT NULL, Flink rejects the conversion. Declare the columns, primary key, and watermark you want to keep, because anything the query leaves out is not carried over. The safe way is to run SHOW CREATE TABLE on the existing table first and paste its exact column list and constraints into the CREATE OR ALTER. Declaring the schema this way keeps the schema already registered for the topic in Schema Registry unchanged, because the statement reuses it instead of letting the query redefine it. Letting the query define the schema risks writing a different schema to Schema Registry and changing the contract that consumers rely on. This risk is highest for an inferred table whose schema lives only in Schema Registry. You can omit the columns only when the schema the query produces already matches the topic. For why declaring the schema explicitly matters, see Explicit and inferred schemas.
Processing starts fresh. The materialized table runs a new query with empty state. It does not resume from the old table’s job, so the RESUME_OR_* start modes fall back to their plain start point. A stateful query, such as a GROUP BY, rebuilds its results by reprocessing the source. Use START_MODE to control how much history is reprocessed.

How adoption treats the existing writer

Adopting a table acts on the Kafka topic behind it. It does not take over whatever is currently writing into that topic. A running statement is a separate object with its own processing state and its own read position, and today you cannot move a running statement into a materialized table.

For example, suppose a long-running statement keeps user_spending up to date:

INSERT INTO user_spending
SELECT customer_id, SUM(price) AS total_spend
FROM examples.marketplace.orders
GROUP BY customer_id;

When you adopt user_spending as a materialized table, Flink starts a new managed query for it. The original INSERT INTO statement is untouched:

It keeps running until you stop it, so two jobs write to the same topic.
Its aggregation state, the running SUM per customer, is not handed to the materialized table. The new query starts with empty state.
Its read position in the source is not carried over. The new query reads according to START_MODE, so FROM_NOW can miss rows written during the switch and FROM_BEGINNING reprocesses everything.

Recommended approach: stop the statement that writes to the topic first, then run CREATE OR ALTER MATERIALIZED TABLE to adopt it.

If you adopt without stopping the old writer, both jobs produce into the same topic at once, with these side effects:

Wrong results on an upsert table. The two jobs write rows for the same keys from different positions and overwrite each other, so the values are inconsistent.
Duplicate rows on an append-only table. Both jobs append the same results.
Double the cost. You pay for two jobs, each consuming at least one Confluent Flink Unit (CFU), until you stop the old one.

Adoption puts a materialized table on a topic that already exists. It is not a way to migrate a running pipeline.

A few more things to know when you adopt an existing table:

Nullability and types must match, not only column names. A common rejection is a column the topic stores as nullable that the query produces as NOT NULL, for example SUM over a NOT NULL source column. Re-declare the column with the same nullability the topic already has. SHOW CREATE TABLE prints exactly what to match.
Offsets are not carried over from a previous statement. You can’t continue from the exact point where a stopped statement left off. A full reprocess repeats rows already in the topic and FROM_NOW can miss data written during the gap. Adoption is not a clean migration path for an existing statement. Use it to put a materialized table on a topic that already exists, not to move a running statement.
Adoption is one-way. After a table becomes a materialized table, you can’t turn it back into a regular table. You can still use it like a regular table by suspending the managed job with ALTER MATERIALIZED TABLE ... SUSPEND and then writing to it with INSERT INTO.
A view can’t be adopted. A view has no backing topic, so there is nothing to adopt.
If the new query can’t start, you get a materialized table with no query attached. Fix the issue and run CREATE OR ALTER MATERIALIZED TABLE again or write to the table with INSERT INTO.

START_MODE

The START_MODE clause controls how much historical data is processed when a materialized table is created or evolved. It answers the question: “What data should the new query logic process?”

If START_MODE is omitted, the default is RESUME_OR_FROM_BEGINNING.

Value	Description
`FROM_BEGINNING`	Reprocesses all available data from the source, starting from the earliest available offset.
`FROM_NOW`	Processes only new data arriving after the command is executed. No historical data is reprocessed.
`FROM_TIMESTAMP(TIMESTAMP '...')`	Reprocesses data starting from a specific absolute timestamp. Specify the timestamp as a `TIMESTAMP` literal, which requires the `TIMESTAMP` keyword and the `yyyy-MM-dd HH:mm:ss` format. For example: `FROM_TIMESTAMP(TIMESTAMP '2026-03-01 00:00:00')`.
`FROM_NOW(INTERVAL '...')`	Reprocesses data from a relative time in the past, evaluated at the time the command is executed. Specify the value as an `INTERVAL` literal, which requires the `INTERVAL` keyword. For example: `FROM_NOW(INTERVAL '7' DAY)`.
`RESUME_OR_FROM_BEGINNING` (default)	On ALTER (existing table): Attempts to resume from the previous job’s last position. On CREATE (new table): Falls back to `FROM_BEGINNING` and reprocesses all available data.
`RESUME_OR_FROM_NOW`	On ALTER: Attempts to resume from the previous job’s last position. On CREATE: Falls back to `FROM_NOW` and processes only new data.
`RESUME_OR_FROM_TIMESTAMP(TIMESTAMP '...')`	On ALTER: Attempts to resume from the previous job’s last position. On CREATE: Falls back to `FROM_TIMESTAMP`.
`RESUME_OR_FROM_NOW(INTERVAL '...')`	On ALTER: Attempts to resume from the previous job’s last position. On CREATE: Falls back to `FROM_NOW(INTERVAL '<interval>')`.

Behavior on CREATE vs ALTER

The RESUME_OR_* options behave differently depending on whether the materialized table is being created for the first time or being evolved:

On CREATE: No previous job exists, so the RESUME part has no effect. The system uses the fallback behavior (FROM_BEGINNING, FROM_NOW, FROM_TIMESTAMP, or FROM_NOW(INTERVAL '<interval>')).
On ALTER: The system first attempts to resume from the previous job’s savepoint, starting exactly where the old job stopped. If resume is not possible, for example, because the query change is incompatible, it falls back to the specified mode.

Interaction with source retention

For any option that reprocesses historical data, like FROM_BEGINNING or FROM_TIMESTAMP, if the specified start point is older than the earliest data available in the source, for example, due to topic retention policies, the job starts processing from the earliest available data point. No error is raised.

Examples

Add a column to an existing materialized table

CREATE OR ALTER MATERIALIZED TABLE enriched_orders (
  `order_id` STRING,
  `customer_id` INT,
  `price` DOUBLE,
  `product_id` STRING
) AS
SELECT order_id, customer_id, price, product_id
FROM examples.marketplace.orders;

The output topic schema is updated to include product_id. Data is reprocessed based on the START_MODE (default: RESUME_OR_FROM_BEGINNING).

Change query logic

Add a filter to an existing materialized table:

CREATE OR ALTER MATERIALIZED TABLE enriched_orders (
  `order_id` STRING,
  `customer_id` INT,
  `price` DOUBLE,
  `product_id` STRING
) AS
SELECT order_id, customer_id, price, product_id
FROM examples.marketplace.orders
WHERE price > 25.00;

Use START_MODE for full reprocessing

Force a full reprocessing of all available historical data:

CREATE OR ALTER MATERIALIZED TABLE enriched_orders
START_MODE = FROM_BEGINNING
AS
SELECT order_id, customer_id, price
FROM examples.marketplace.orders;

Use START_MODE with a relative interval

Reprocess only the last seven days of data:

CREATE OR ALTER MATERIALIZED TABLE enriched_orders
START_MODE = FROM_NOW(INTERVAL '7' DAY)
AS
SELECT order_id, customer_id, price
FROM examples.marketplace.orders;

Downstream consumer considerations

Because evolutions discard state and rebuild from scratch, downstream consumers might observe specific behaviors. For detailed guidance, see Downstream consumer impact.

Upsert mode: Keys that are filtered out by the new query logic have no delete message emitted. Old records remain downstream as stale “zombie” data.
Append-only mode: Reprocessed records appear alongside the original records, resulting in duplicates in the output topic.
Retract mode: The new job emits INSERT messages for the reprocessed data but does not emit DELETE messages for the old result set.

Limitations

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