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

# ALTER MATERIALIZED TABLE Statement in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables modifying an existing materialized table by using the
`ALTER MATERIALIZED TABLE` statement. You can change table properties,
add columns, or evolve the continuous query.

## Syntax

```sql
ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name {
   ADD (metadata_column_name metadata_column_type METADATA [FROM metadata_key] VIRTUAL [COMMENT column_comment])
 | ADD (computed_column_name AS computed_column_expression [COMMENT column_comment])
 | MODIFY WATERMARK FOR rowtime_column_name AS watermark_strategy_expression
 | DROP WATERMARK
 | SET (key1='value1' [, key2='value2', ...])
 | RESET (key1 [, key2, ...])
 | SUSPEND
}

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

`ALTER MATERIALIZED TABLE` supports three kinds of operations:

### Property changes

The following changes modify table metadata without triggering an evolution
of the continuous query:

- Add [metadata columns](create-table.md#flink-sql-metadata-columns)
- Add computed columns
- Change or remove the [watermark](../../concepts/timely-stream-processing.md#flink-sql-event-time-and-watermarks)
- Modify [table properties](create-table.md#flink-sql-with-options)

Physical columns cannot be added, modified, or dropped within Flink directly,
but schemas can be
[evolved in Schema Registry](../../../sr/fundamentals/schema-evolution.md#schema-evolution-and-compatibility).

### Suspend the refresh pipeline

`SUSPEND` pauses the continuous query that refreshes the materialized
table, without dropping the table or its backing Kafka topic. Use `SUSPEND`
to stop processing temporarily, for example, while investigating an issue
or performing maintenance on upstream sources.

To resume a stopped materialized table, use the
[REST API](../../operate-and-deploy/flink-rest-api.md#flink-rest-api) or the `stopped` attribute of the
[Confluent Terraform provider](../../operate-and-deploy/terraform.md#flink-sql-terraform). For more
information, see
[Lifecycle operations](../../concepts/materialized-tables.md#flink-sql-materialized-tables-lifecycle).

### Query evolution

`ALTER MATERIALIZED TABLE` can also change the `AS SELECT` query,
triggering a full evolution of the materialized table. This works the same
way as
[CREATE OR ALTER MATERIALIZED TABLE](create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table),
except that the materialized table must already exist.

When an evolution is triggered:

- The existing continuous query is stopped.
- All Flink processing state is discarded.
- A new continuous query starts with the updated query.
- Results are written to the same output topic.
- The `START_MODE` clause controls data reprocessing behavior.

For the full list of `START_MODE` values, see
[START_MODE](create-or-alter-materialized-table.md#flink-sql-start-mode).

## Examples

Add a metadata column to expose Apache Kafka® timestamps:

```sql
ALTER MATERIALIZED TABLE enriched_orders
ADD (`kafka_ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL);
```

Add a computed column:

```sql
ALTER MATERIALIZED TABLE enriched_orders
ADD (`price_with_vat` AS price * 1.21);
```

Change a table property:

```sql
ALTER MATERIALIZED TABLE enriched_orders
SET ('kafka.retention.time' = '30 d');
```

Add a watermark definition:

```sql
ALTER MATERIALIZED TABLE enriched_orders
MODIFY WATERMARK FOR `event_time` AS `event_time` - INTERVAL '5' SECOND;
```

Suspend the refresh pipeline:

```sql
ALTER MATERIALIZED TABLE enriched_orders SUSPEND;
```

Evolve the query:

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

## Related content

- [CREATE OR ALTER MATERIALIZED TABLE](create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table)
- [CREATE MATERIALIZED TABLE](create-materialized-table.md#flink-sql-create-materialized-table)
- [DROP MATERIALIZED TABLE](drop-materialized-table.md#flink-sql-drop-materialized-table)
- [ALTER TABLE](alter-table.md#flink-sql-alter-table)
- [Schema and Statement Evolution](../../concepts/schema-statement-evolution.md#flink-sql-schema-and-statement-evolution)

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