# 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](create-materialized-table.md#flink-sql-create-materialized-table). ## Syntax ```sql CREATE OR ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name [( { | | }[ , ...n] [ ] [ ][ , ...n] )] [COMMENT table_comment] [DISTRIBUTED BY (column_name1, column_name2, ...) INTO n BUCKETS] [WITH (key1=value1, key2=value2, ...)] [START_MODE = ] AS ``` ## 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](create-materialized-table.md#flink-sql-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](#flink-sql-adopt-existing-table). ### How evolution works An evolution performs an in-place migration: 1. The existing continuous query is stopped. 2. A new continuous query is created with the updated query, schema, and configuration. 3. The new query begins processing data according to the `START_MODE` setting. 4. 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](create-materialized-table.md#flink-sql-mt-explicit-schema). 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: ```sql SHOW CREATE TABLE user_spending; ``` ```sql 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](create-materialized-table.md#flink-sql-mt-explicit-schema). - **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](#flink-sql-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: ```sql 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](#flink-sql-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 '')`. | ### 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 '')`). - **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 ```sql 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: ```sql 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: ```sql 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: ```sql 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](../../concepts/materialized-tables.md#flink-sql-materialized-tables-downstream-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](../../concepts/materialized-tables.md#flink-sql-materialized-tables-limitations). ## Related content - [CREATE MATERIALIZED TABLE](create-materialized-table.md#flink-sql-create-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) - [Schema and Statement Evolution](../../concepts/schema-statement-evolution.md#flink-sql-schema-and-statement-evolution) - [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).