<a id="flink-sql-examples"></a>

# Flink SQL Examples in Confluent Cloud for Apache Flink

The following examples show common Flink SQL patterns in
Confluent Cloud for Apache Flink®, including creating tables backed by Apache Kafka® topics,
working with [inferred tables](statements/create-table.md#flink-sql-create-table-inferred-tables)
that Flink derives automatically from Kafka topics and
[Schema Registry](../../sr/schemas-manage.md#sr-prv) schemas, altering table properties,
running SELECT queries, and using schema references.

- [CREATE TABLE](#flink-sql-examples-create-table)
- [Inferred tables](#flink-sql-examples-inferred-tables)
- [ALTER TABLE](#flink-sql-examples-alter-table)
- [SELECT](#flink-sql-examples-select)
- [Schema reference](#flink-sql-examples-schema-reference)

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

## CREATE TABLE examples

The following examples show how to create Flink tables with various options.

### Minimal table

The smallest valid CREATE TABLE statement declares a single column and
relies on default settings for changelog mode, distribution, and partitions.

```sql
CREATE TABLE t_minimal (s STRING);
```

Properties
: - Append changelog mode.
  - No Schema Registry key.
  - Round-robin distribution.
  - 6 Kafka partitions.
  - The `$rowtime` column and system watermark are added implicitly.

### Table with a primary key

Syntax
: ```sql
  CREATE TABLE t_pk (k INT PRIMARY KEY NOT ENFORCED, s STRING);
  ```

Properties
: - Upsert changelog mode.
  - The primary key defines an implicit DISTRIBUTED BY(k).
  - `k` is the Schema Registry key.
  - Hash distribution on `k`.
  - The table has 6 Kafka partitions.
  - `k` is declared as being unique, meaning no duplicate rows.
  - `k` must not contain NULLs, so an implicit NOT NULL is added.
  - The `$rowtime` column and system watermark are added implicitly.

### Table with a primary key in append mode

Syntax
: ```sql
  CREATE TABLE t_pk_append (k INT PRIMARY KEY NOT ENFORCED, s STRING)
    DISTRIBUTED INTO 4 BUCKETS
    WITH ('changelog.mode' = 'append');
  ```

Properties
: - Append changelog mode.
  - `k` is the Schema Registry key.
  - Hash distribution on `k`.
  - The table has 4 Kafka partitions.
  - `k` is declared as being unique, meaning no duplicate rows.
  - `k` must not contain NULLs, meaning implicit NOT NULL.
  - The `$rowtime` column and system watermark are added implicitly.

### Table with hash distribution

Syntax
: ```sql
  CREATE TABLE t_dist (k INT, s STRING) DISTRIBUTED BY (k) INTO 4 BUCKETS;
  ```

Properties
: - Append changelog mode.
  - `k` is the Schema Registry key.
  - Hash distribution on `k`.
  - The table has 4 Kafka partitions.
  - The `$rowtime` column and system watermark are added implicitly.

### Complex table with all concepts combined

Syntax
: ```sql
  CREATE TABLE t_complex (k1 INT, k2 INT, PRIMARY KEY (k1, k2) NOT ENFORCED, s STRING)
    COMMENT 'My complex table'
    DISTRIBUTED BY HASH(k1) INTO 4 BUCKETS
    WITH ('changelog.mode' = 'append');
  ```

Properties
: - Append changelog mode.
  - `k1` is the Schema Registry key.
  - Hash distribution on `k1`.
  - `k2` is treated as a value column and is stored in the value part of Schema Registry.
  - The table has 4 Kafka partitions.
  - `k1` and `k2` are declared as being unique, meaning no duplicates.
  - `k` and `k2` must not contain NULLs, meaning implicit NOT NULL.
  - The `$rowtime` column and system watermark are added implicitly.
  - An additional comment is added.

### Table with overlapping names in key/value of Schema Registry but disjoint data

Syntax
: ```sql
  CREATE TABLE t_disjoint (from_key_k INT, k STRING)
    DISTRIBUTED BY (from_key_k)
    WITH ('key.fields-prefix' = 'from_key_');
  ```

Properties
: - Append changelog mode.
  - Hash distribution on `from_key_k`.
  - The key prefix `from_key_` is defined and is stripped before storing the
    schema in Schema Registry.
    - Therefore, `k` is the Schema Registry key of type INT.
    - Also, `k` is the Schema Registry value of type STRING.
  - Both key and value store disjoint data, so they can have different data types.

### Table with overlapping names in key/value of Schema Registry but joint data

Syntax
: ```sql
  CREATE TABLE t_joint (k INT, v STRING)
    DISTRIBUTED BY (k)
    WITH ('value.fields-include' = 'all');
  ```

Properties
: - Append changelog mode.
  - Hash distribution on `k`.
  - By default, the key is never included in the value in Schema Registry.
  - By setting `'value.fields-include' = 'all'`, the value contains the full table schema.
    - Therefore, `k` is the Schema Registry key.
    - Also, `k, v` is the Schema Registry value.
  - The payload of `k` is stored twice in the Kafka message, because key and
    value store joint data and they have the same data type for `k`.

### Table with metadata columns for writing a Kafka message timestamp

Syntax
: ```sql
  CREATE TABLE t_metadata_write (name STRING, ts TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp')
    DISTRIBUTED INTO 1 BUCKETS;
  ```

Properties
: - Adds the `ts` metadata column, which isn’t part of Schema Registry but instead is a
    pure Flink concept.
  - In contrast with `$rowtime`, which is declared as a METADATA VIRTUAL column,
    `ts` is selected in a SELECT \* statement and is writable.

The following examples show how to fill Kafka messages with an
[instant](datatypes.md#flink-sql-timestamp-comparison-timestamp-ltz).

```mysql
INSERT INTO t (ts, name) SELECT NOW(), 'Alice';
INSERT INTO t (ts, name) SELECT TO_TIMESTAMP_LTZ(0, 3), 'Bob';
SELECT $rowtime, * FROM t;
```

The Schema Registry subject compatibility mode must be FULL or FULL_TRANSITIVE.
For more information, see [Schema Evolution and Compatibility for Schema Registry on Confluent Cloud](../../sr/fundamentals/schema-evolution.md#schema-evolution-and-compatibility).

### Table with string key and value in Schema Registry

Syntax
: ```sql
  CREATE TABLE t_raw_string_key (key STRING, i INT)
    DISTRIBUTED BY (key)
    WITH ('key.format' = 'raw');
  ```

Properties
: - Schema Registry is filled with a value subject containing `i`.
  - The key columns are determined by the DISTRIBUTED BY clause.
  - By default, Avro in Schema Registry would be used for the key, but the WITH clause
    overrides this to the `raw` format.

### Tables with cross-region schema sharing

1. Create two Kafka clusters in different regions, for example, `eu-west-1` and
   `us-west-2`.
2. Create two Flink compute pools in different regions, for example,
   `eu-west-1` and `us-west-2`.
3. In the first region, run the following statement.
   ```sql
   CREATE TABLE t_shared_schema (key STRING, s STRING) DISTRIBUTED BY (key);
   ```
4. In the second region, run the same statement.
   ```sql
   CREATE TABLE t_shared_schema (key STRING, s STRING) DISTRIBUTED BY (key);
   ```

Properties
: - Schema Registry is shared across regions.
  - The SQL metastore, Flink compute pools, and Kafka clusters are regional.
  - Both tables in either region share the Schema Registry subjects `t_shared_schema-key`
    and `t_shared_schema-value`.

### Tables with different changelog modes

There are three ways of storing events in a table’s log, that is, in the
underlying Kafka topic.

append
: - Every insertion event is an **immutable fact**.
  - Every event is **insert-only**.
  - Events can be distributed in a round-robin fashion across workers/shards
    because they are **unrelated**.

upsert
: - Events are **related** using a primary key.
  - Every event is either an **upsert or delete** event for a primary key.
  - Events for the same primary key should land at the same worker/shard.

retract
: - Every upsert event is a **fact that can be “undone”**.
  - This means that every event is either an insertion or its retraction.
  - So, **two events are related by all columns**. In other words, the entire
    row is the key.
  <br/>
    For example, `+I['Bob', 42]` is related to `-D['Bob', 42]` and
    `+U['Alice', 13]` is related to `-U['Alice', 13]`.

- The **retract** mode is intermediate between the **append** and **upsert**
  modes.
- The **append** and **upsert** modes are natural to existing Kafka consumers
  and producers.
- Kafka compaction is a kind of **upsert**.

Start with a table created by the following statement.

```sql
CREATE TABLE t_changelog_modes (i BIGINT);
```

Properties
: - Confluent Cloud for Apache Flink always derives an appropriate changelog mode for the preceding
    declaration.
  - If there is no primary key, **append** is the safest option, because it
    prevents users from pushing updates into a topic accidentally, and it has
    the best support of downstream consumers.
  <br/>
  ```sql
  -- works because the query is non-updating
  INSERT INTO t_changelog_modes SELECT 1;
  <br/>
  -- does not work because the query is updating, causing an error
  INSERT INTO t_changelog_modes SELECT COUNT(*) FROM (VALUES (1), (2), (3));
  ```

If you need updates, and if downstream consumers support it, for example, when
the consumer is another Flink job, you can set the changelog mode to **retract**.

```sql
ALTER TABLE t_changelog_modes SET ('changelog.mode' = 'retract');
```

Properties
: - The table starts accepting retractions during INSERT INTO.
  - Already existing records in the Kafka topic are treated as insertions.
  - Newly added records receive a changeflag (+I, +U, -U, -D) in the Kafka
    message header.
  <br/>
  Going back to **append** mode is possible, but retractions (-U, -D) appear
  as insertions, and the Kafka header metadata column reveals the changeflag.
  <br/>
  ```sql
  ALTER TABLE t_changelog_modes SET ('changelog.mode' = 'append');
  ALTER TABLE t_changelog_modes ADD headers MAP<BYTES, BYTES> METADATA VIRTUAL;
  <br/>
  -- Shows what is serialized internally
  SELECT i, headers FROM t_changelog_modes;
  ```

### Table with infinite retention time

```sql
CREATE TABLE t_infinite_retention (i INT) WITH ('kafka.retention.time' = '0');
```

Properties
: - By default, the retention time is 7 days, as in all other APIs.
  - Flink doesn’t support `-1` for durations, so `0` means infinite
    retention time.
  - Durations in Flink support `2 day` or `2 d` syntax, so it doesn’t need
    to be in milliseconds.
  - If no unit is specified, the unit is milliseconds.
  - The following units are supported:
  <br/>
  ```text
  "d", "day", "h", "hour", "m", "min", "minute", "ms", "milli", "millisecond",
  "µs", "micro", "microsecond", "ns", "nano", "nanosecond"
  ```

<a id="flink-sql-examples-inferred-tables"></a>

## Inferred table examples

<!-- flink-sql-inferred-table-example_start -->

Inferred tables are tables that have not been created by using a CREATE TABLE
statement, but instead are automatically detected from information about
existing Kafka topics and Schema Registry entries.

You can use the ALTER TABLE statement to
[evolve schemas](statements/alter-table.md#flink-sql-alter-table-examples) for inferred
tables.

The following examples show output from the SHOW CREATE TABLE statement called
on the resulting table.

### No key or value in Schema Registry

For an inferred table with no registered key or value schemas, SHOW CREATE TABLE
returns the following output:

```sql
CREATE TABLE `t_raw` (
  `key` VARBINARY(2147483647),
  `val` VARBINARY(2147483647)
) DISTRIBUTED BY HASH(`key`) INTO 2 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'key.format' = 'raw',
  'value.format' = 'raw'
  ...
)
```

Properties
: - Key and value formats are raw (binary format) with BYTES.
  - Following Kafka message semantics, both key and value also support NULL,
    so the following code is valid:
    ```sql
    INSERT INTO t_raw (key, val) SELECT CAST(NULL AS BYTES), CAST(NULL AS BYTES);
    ```

### No key but record value in Schema Registry

For the following value schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "i",
      "type": "int"
    },
    {
      "name": "s",
      "type": "string"
    }
  ]
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_raw_key` (
  `key` VARBINARY(2147483647),
  `i` INT NOT NULL,
  `s` VARCHAR(2147483647) NOT NULL
) DISTRIBUTED BY HASH(`key`) INTO 6 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'key.format' = 'raw',
  'value.format' = 'avro-registry'
  ...
)
```

Properties
: - The key format is raw (binary format) with BYTES.
  - Following Kafka message semantics, the key also supports NULL, so the
    following code is valid:
    ```sql
    INSERT INTO t_raw_key SELECT CAST(NULL AS BYTES), 12, 'Bob';
    ```

### Atomic key and record value in Schema Registry

For the following key schema in Schema Registry:

```text
"int"
```

And for the following value schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "i",
      "type": "int"
    },
    {
      "name": "s",
      "type": "string"
    }
  ]
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_atomic_key` (
  `key` INT NOT NULL,
  `i` INT NOT NULL,
  `s` VARCHAR(2147483647) NOT NULL
) DISTRIBUTED BY HASH(`key`) INTO 2 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'key.format' = 'avro-registry',
  'value.format' = 'avro-registry'
  ...
)
```

Properties
: - Schema Registry defines the column data type as INT NOT NULL.
  - The column name, `key`, is used as the default, because Schema Registry doesn’t
    provide a column name.

### Overlapping names in key/value, no key in Schema Registry

For the following value schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "i",
      "type": "int"
    },
    {
      "name": "key",
      "type": "string"
    }
  ]
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_raw_disjoint` (
  `key_key` VARBINARY(2147483647),
  `i` INT NOT NULL,
  `key` VARCHAR(2147483647) NOT NULL
) DISTRIBUTED BY HASH(`key_key`) INTO 1 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'key.fields-prefix' = 'key_',
  'key.format' = 'raw',
  'value.format' = 'avro-registry'
  ...
)
```

Properties
: - The Schema Registry value schema defines columns `i INT NOT NULL` and `key STRING`.
  - The column name `key BYTES` is used as the default if no key is in Schema Registry.
  - Because `key` would collide with value schema column, the `key_` prefix
    is added.

### Record key and record value in Schema Registry

For the following key schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "uid",
      "type": "int"
    }
  ]
}
```

And for the following value schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "zip_code",
      "type": "string"
    }
  ]
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_sr_disjoint` (
  `uid` INT NOT NULL,
  `name` VARCHAR(2147483647) NOT NULL,
  `zip_code` VARCHAR(2147483647) NOT NULL
) DISTRIBUTED BY HASH(`uid`) INTO 1 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'value.format' = 'avro-registry'
  ...
)
```

Properties
: - Schema Registry defines columns for both key and value.
  - The column names of key and value are disjoint sets and don’t overlap.

### Record key and record value with overlap in Schema Registry

For the following key schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "uid",
      "type": "int"
    }
  ]
}
```

And for the following value schema in Schema Registry:

```json
{
    "type": "record",
    "name": "TestRecord",
    "fields": [
      {
        "name": "uid",
        "type": "int"
      },{
        "name": "name",
        "type": "string"
      },
      {
        "name": "zip_code",
        "type": "string"
      }
    ]
  }
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_sr_joint` (
  `uid` INT NOT NULL,
  `name` VARCHAR(2147483647) NOT NULL,
  `zip_code` VARCHAR(2147483647) NOT NULL
) DISTRIBUTED BY HASH(`uid`) INTO 1 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'value.fields-include' = 'all',
  'value.format' = 'avro-registry'
  ...
)
```

Properties
: - Schema Registry defines columns for both key and value.
  - The column names of key and value overlap on `uid`.
  - `'value.fields-include' = 'all'` is set to exclude the key, because it
    is fully contained in the value.
  - Detecting that key is fully contained in the value requires that
    *both field name and data type match completely, including nullability*,
    and *all fields of the key* are included in the value.

### Union types in Schema Registry

For the following value schema in Schema Registry:

```text
["int", "string"]
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_union` (
  `key` VARBINARY(2147483647),
  `int` INT,
  `string` VARCHAR(2147483647)
)
...
```

For the following value schema in Schema Registry:

```json
[
  "string",
  {
    "type": "record",
    "name": "User",
    "fields": [
      {
        "name": "uid",
        "type": "int"
      },{
        "name": "name",
        "type": "string"
      }
    ]
  },
  {
    "type": "record",
    "name": "Address",
    "fields": [
      {
        "name": "zip_code",
        "type": "string"
      }
    ]
  }
]
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_union` (
  `key` VARBINARY(2147483647),
  `string` VARCHAR(2147483647),
  `User` ROW<`uid` INT NOT NULL, `name` VARCHAR(2147483647) NOT NULL>,
  `Address` ROW<`zip_code` VARCHAR(2147483647) NOT NULL>
)
...
```

Properties
: - NULL and NOT NULL are inferred depending on whether a union contains
    NULL.
  - Elements of a union are always NULL, because they need to be set to NULL
    when a different element is set.
  - If a record defines a `namespace`, the field is prefixed with it,
    for example, `org.myorg.avro.User`.

### Multi-message protobuf schema in Schema Registry

For the following value schema in Schema Registry:

```protobuf
syntax = "proto3";

message Purchase {
   string item = 1;
   double amount = 2;
   string customer_id = 3;
}

message Pageview {
   string url = 1;
   bool is_special = 2;
   string customer_id = 3;
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t` (
  `key` VARBINARY(2147483647),
  `Purchase` ROW<
      `item` VARCHAR(2147483647) NOT NULL,
      `amount` DOUBLE NOT NULL,
      `customer_id` VARCHAR(2147483647) NOT NULL
   >,
  `Pageview` ROW<
      `url` VARCHAR(2147483647) NOT NULL,
      `is_special` BOOLEAN NOT NULL,
      `customer_id` VARCHAR(2147483647) NOT NULL
   >
)
...
```

For the following value schema in Schema Registry:

```protobuf
syntax = "proto3";

message Purchase {
   string item = 1;
   double amount = 2;
   string customer_id = 3;
   Pageview pageview = 4;
}

message Pageview {
   string url = 1;
   bool is_special = 2;
   string customer_id = 3;
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t` (
  `key` VARBINARY(2147483647),
  `Purchase` ROW<
      `item` VARCHAR(2147483647) NOT NULL,
      `amount` DOUBLE NOT NULL,
      `customer_id` VARCHAR(2147483647) NOT NULL,
      `pageview` ROW<
         `url` VARCHAR(2147483647) NOT NULL,
         `is_special` BOOLEAN NOT NULL,
         `customer_id` VARCHAR(2147483647) NOT NULL
      >
   >,
  `Pageview` ROW<
      `url` VARCHAR(2147483647) NOT NULL,
      `is_special` BOOLEAN NOT NULL,
      `customer_id` VARCHAR(2147483647) NOT NULL
   >
)
...
```

For the following value schema in Schema Registry:

```protobuf
syntax = "proto3";

message Purchase {
   string item = 1;
   double amount = 2;
   string customer_id = 3;
   Pageview pageview = 4;
   message Pageview {
      string url = 1;
      bool is_special = 2;
      string customer_id = 3;
   }
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t` (
  `key` VARBINARY(2147483647),
  `item` VARCHAR(2147483647) NOT NULL,
  `amount` DOUBLE NOT NULL,
  `customer_id` VARCHAR(2147483647) NOT NULL,
  `pageview` ROW<
      `url` VARCHAR(2147483647) NOT NULL,
      `is_special` BOOLEAN NOT NULL,
      `customer_id` VARCHAR(2147483647) NOT NULL
   >
)
...
```

### Debezium CDC format in Schema Registry

For a Debezium CDC format with the following value schema in Schema Registry:

```json
{
  "type": "record",
  "name": "Customer",
  "namespace": "io.debezium.data",
  "fields": [
    {
      "name": "before",
      "type": ["null", {
        "type": "record",
        "name": "Value",
        "fields": [
          {"name": "id", "type": "int"},
          {"name": "name", "type": "string"},
          {"name": "email", "type": "string"}
        ]
      }],
      "default": null
    },
    {
      "name": "after",
      "type": ["null", "Value"],
      "default": null
    },
    {
      "name": "source",
      "type": {
        "type": "record",
        "name": "Source",
        "fields": [
          {"name": "version", "type": "string"},
          {"name": "connector", "type": "string"},
          {"name": "name", "type": "string"},
          {"name": "ts_ms", "type": "long"},
          {"name": "db", "type": "string"},
          {"name": "schema", "type": "string"},
          {"name": "table", "type": "string"}
        ]
      }
    },
    {"name": "op", "type": "string"},
    {"name": "ts_ms", "type": ["null", "long"], "default": null},
    {"name": "transaction", "type": ["null", {
      "type": "record",
      "name": "Transaction",
      "fields": [
        {"name": "id", "type": "string"},
        {"name": "total_order", "type": "long"},
        {"name": "data_collection_order", "type": "long"}
      ]
    }], "default": null}
  ]
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `customer_changes` (
  `key` VARBINARY(2147483647),
   `id` INT NOT NULL,
   `name` VARCHAR(2147483647) NOT NULL,
   `email` VARCHAR(2147483647) NOT NULL
)
DISTRIBUTED BY HASH(`key`) INTO 6 BUCKETS
WITH (
  'changelog.mode' = 'retract',
  'connector' = 'confluent',
  'key.format' = 'raw',
  'value.format' = 'avro-debezium-registry'
  ...
)
```

Properties
: - Flink detects the Debezium format automatically, based on the schema
    structure with `after`, `before`, and `op` fields.
  - The table schema is inferred from the `after` schema, exposing only the
    actual data fields.
  - For a full explanation of Debezium format support and changelog modes,
    see [Debezium format](serialization.md#flink-sql-serialization-debezium-format).
  - **Automatic Debezium envelope detection**: For schemas created after
    May 19, 2025 at 09:00 UTC, Flink automatically detects Debezium envelopes
    and sets appropriate defaults:
    * `value.format` defaults to `*-debezium-registry` (instead of
      `*-registry`).
    * `changelog.mode` defaults to `retract` (instead of `append`).
    * Exception: If Kafka `cleanup.policy` is `compact`, Flink sets
      `changelog.mode` to `upsert`.
  - The default `changelog.mode` is `retract`, which properly handles all
    CDC operations, including inserts, updates, and deletes.
  - You can manually override the changelog mode if necessary:
    ```sql
    -- Change to upsert mode for primary key-based operations
    ALTER TABLE customer_changes SET ('changelog.mode' = 'upsert');
  <br/>
    -- Change to append mode (processes only inserts and updates)
    ALTER TABLE customer_changes SET ('changelog.mode' = 'append');
    ```

<!-- flink-sql-inferred-table-example_end -->

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

## ALTER TABLE examples

The following examples show frequently used scenarios for ALTER TABLE.

### Define a watermark for perfectly ordered data

Flink guarantees that rows are always emitted before the watermark is generated.
The following statements ensure that for perfectly ordered events, meaning
events without time-skew, a watermark can be equal to the timestamp or 1 ms less than
the timestamp.

```mysql
CREATE TABLE t_perfect_watermark (i INT);

-- If multiple events can have the same timestamp.
ALTER TABLE t_perfect_watermark
  MODIFY WATERMARK FOR $rowtime AS $rowtime - INTERVAL '0.001' SECOND;

-- If a single event can have the timestamp.
ALTER TABLE t_perfect_watermark
  MODIFY WATERMARK FOR $rowtime AS $rowtime;
```

### Drop your custom watermark strategy

Remove the custom watermark strategy to restore the
[default watermark strategy](statements/create-table.md#flink-sql-watermark-clause).

1. View the current table schema and metadata.
   ```sql
   DESCRIBE `orders`;
   ```

   Your output should resemble:
   ```none
   +-------------+------------------------+----------+-------------------+
   | Column Name |       Data Type        | Nullable |      Extras       |
   +-------------+------------------------+----------+-------------------+
   | user        | BIGINT                 | NOT NULL | PRIMARY KEY       |
   | product     | STRING                 | NULL     |                   |
   | amount      | INT                    | NULL     |                   |
   | ts          | TIMESTAMP(3) *ROWTIME* | NULL     | WATERMARK AS `ts` |
   +-------------+------------------------+----------+-------------------+
   ```
2. Remove the watermark strategy of the table.
   ```sql
   ALTER TABLE `orders` DROP WATERMARK;
   ```

   Your output should resemble:
   ```none
   Statement phase is COMPLETED.
   ```
3. Check the new table schema and metadata.
   ```sql
   DESCRIBE `orders`;
   ```

   Your output should resemble:
   ```text
   +-------------+--------------+----------+-------------+
   | Column Name |  Data Type   | Nullable |   Extras    |
   +-------------+--------------+----------+-------------+
   | user        | BIGINT       | NOT NULL | PRIMARY KEY |
   | product     | STRING       | NULL     |             |
   | amount      | INT          | NULL     |             |
   | ts          | TIMESTAMP(3) | NULL     |             |
   +-------------+--------------+----------+-------------+
   ```

### Add or drop a PRIMARY KEY constraint

Add a named `PRIMARY KEY` constraint to an existing table, or remove one by
name.

```sql
-- Create an example table with a non-nullable id column.
CREATE TABLE t_pk_example (id INT NOT NULL, name STRING);

-- Add a primary key constraint on the id column.
ALTER TABLE t_pk_example ADD CONSTRAINT pk_example PRIMARY KEY (id) NOT ENFORCED;

-- Remove the constraint by name.
ALTER TABLE t_pk_example DROP CONSTRAINT pk_example;
```

Properties
: - Name the constraint explicitly when you add it. The
    `CONSTRAINT constraint_name` clause is optional, but dropping the
    constraint later requires knowing its name.
  - No column in a primary key can be nullable, so the column must already
    be declared `NOT NULL`.
  - Flink SQL supports only `NOT ENFORCED` primary key constraints. For
    details, see the
    [PRIMARY KEY constraint](statements/create-table.md#flink-sql-primary-constraint) section of
    the [CREATE TABLE statement](statements/create-table.md#flink-sql-create-table).
  - Adding a `PRIMARY KEY` constraint implicitly distributes the table by
    the key column. This matches the behavior when you declare the
    constraint in a [CREATE TABLE](statements/create-table.md#flink-sql-create-table) statement.

### Configure Debezium format for CDC data

#### Change regular format to Debezium format

#### NOTE
For schemas created after May 19, 2025 at 09:00 UTC, Flink automatically
detects Debezium envelopes and configures the appropriate format and
changelog mode. Manual conversion is necessary only for older schemas or
when you want to override the default behavior.

For a full explanation of how Flink detects the Debezium envelope and chooses
default values for `value.format` and `changelog.mode`, see
[Debezium format](serialization.md#flink-sql-serialization-debezium-format) and
[Changelog modes](serialization.md#flink-sql-changelog-modes).

For tables that have been inferred with regular formats but contain Debezium
CDC (Change Data Capture) data:

### Avro

```sql
-- Convert from regular Avro format to Debezium CDC format
-- and configure the appropriate Flink changelog interpretation mode:
-- * append:  Treats each record as an INSERT operation with no relationship between records
-- * retract: Handles paired operations (INSERT/UPDATE/DELETE) where changes to the same row
--            are represented as a retraction of the old value followed by an addition of the new value
-- * upsert: Groups all operations for the primary key (derived from the Kafka message key),
--           with each operation effectively merging with or replacing previous state
--           (INSERT creates, UPDATE modifies, DELETE removes)
ALTER TABLE customer_data SET (
  'value.format' = 'avro-debezium-registry',
  'changelog.mode' = 'retract'
);
```

### JSON Schema

```sql
-- Convert from regular JSON format to Debezium CDC format
-- and configure the appropriate Flink changelog interpretation mode:
-- * append:  Treats each record as an INSERT operation with no relationship between records
-- * retract: Handles paired operations (INSERT/UPDATE/DELETE) where changes to the same row
--            are represented as a retraction of the old value followed by an addition of the new value
-- * upsert: Groups all operations for the primary key (derived from the Kafka message key),
--           with each operation effectively merging with or replacing previous state
--           (INSERT creates, UPDATE modifies, DELETE removes)
ALTER TABLE customer_data_json SET (
  'value.format' = 'json-debezium-registry',
  'changelog.mode' = 'retract'
);
```

### Protobuf

```sql
-- Convert from regular Protobuf format to Debezium CDC format
-- and configure the appropriate Flink changelog interpretation mode:
-- * append:  Treats each record as an INSERT operation with no relationship between records
-- * retract: Handles paired operations (INSERT/UPDATE/DELETE) where changes to the same row
--            are represented as a retraction of the old value followed by an addition of the new value
-- * upsert: Groups all operations for the primary key (derived from the Kafka message key),
--           with each operation effectively merging with or replacing previous state
--           (INSERT creates, UPDATE modifies, DELETE removes)
ALTER TABLE customer_data_proto SET (
  'value.format' = 'proto-debezium-registry',
  'changelog.mode' = 'retract'
);
```

### Modify changelog processing mode

For tables with any type of data that need a different processing mode
for handling changes:

```sql
-- Change to append mode (default)
-- Best for event streams where each record is independent
ALTER TABLE customer_changes SET (
  'changelog.mode' = 'append'
);

-- Change to retract mode
-- Useful when changes to the same row are represented as paired operations
ALTER TABLE customer_changes SET (
  'changelog.mode' = 'retract'
);

-- Change upsert mode when working with primary keys
-- Best when tracking state changes using a primary key (derived from Kafka message key)
ALTER TABLE customer_changes SET (
  'changelog.mode' = 'upsert'
);
```

### Read and write Kafka headers

```sql
-- Create example topic
CREATE TABLE t_headers (i INT);

-- For read-only (virtual)
ALTER TABLE t_headers ADD headers MAP<BYTES, BYTES> METADATA VIRTUAL;

-- For read and write (persisted). Column becomes mandatory in INSERT INTO.
ALTER TABLE t_headers MODIFY headers MAP<BYTES, BYTES> METADATA;

-- Use implicit casting (origin is always MAP<BYTES, BYTES>)
ALTER TABLE t_headers MODIFY headers MAP<STRING, STRING> METADATA;

-- Insert and read
INSERT INTO t_headers SELECT 42, MAP['k1', 'v1', 'k2', 'v2'];
SELECT * FROM t_headers;
```

Properties
: - The metadata key is `headers`. If you don’t want to name the column this
    way, use: `other_name MAP<BYTES, BYTES> METADATA FROM 'headers' VIRTUAL`.
  - Keys of headers must be unique. Multi-key headers are not supported.

### Add headers as a metadata column

You can get the headers of a Kafka record as a map of raw bytes by adding a
`headers` virtual metadata column.

1. Run the following statement to add the Kafka partition as a metadata column:
   ```sql
   ALTER TABLE `orders` ADD (
     `headers` MAP<BYTES,BYTES> METADATA VIRTUAL);
   ```
2. View the new schema.
   ```sql
   DESCRIBE `orders`;
   ```

   Your output should resemble:
   ```text
   +-------------+-------------------+----------+-------------------------+
   | Column Name |     Data Type     | Nullable |         Extras          |
   +-------------+-------------------+----------+-------------------------+
   | user        | BIGINT            | NOT NULL | PRIMARY KEY, BUCKET KEY |
   | product     | STRING            | NULL     |                         |
   | amount      | INT               | NULL     |                         |
   | ts          | TIMESTAMP(3)      | NULL     |                         |
   | headers     | MAP<BYTES, BYTES> | NULL     | METADATA VIRTUAL        |
   +-------------+-------------------+----------+-------------------------+
   ```

### Read topic from specific offsets

```mysql
-- Create example topic with 1 partition filled with values
CREATE TABLE t_specific_offsets (i INT) DISTRIBUTED INTO 1 BUCKETS;
INSERT INTO t_specific_offsets VALUES (1), (2), (3), (4), (5);

-- Returns 1, 2, 3, 4, 5
SELECT * FROM t_specific_offsets;

-- Changes the scan range
ALTER TABLE t_specific_offsets SET (
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:3'
);

-- Returns 4, 5
SELECT * FROM t_specific_offsets;
```

Properties
: - `scan.startup.mode` and `scan.bounded.mode` control which range in the
    changelog (Kafka topic) to read.
  - `scan.startup.specific-offsets` and `scan.bounded.specific-offsets`
    define offsets per partition.
  - In the example, only 1 partition is used. For multiple partitions, use the
    following syntax:
  <br/>
  ```properties
  'scan.startup.specific-offsets' = 'partition:0,offset:3; partition:1,offset:42; partition:2,offset:0'
  ```

### Debug “no output” and no watermark cases

The root cause for most “no output” cases is that a time-based operation, for
example, TUMBLE, MATCH_RECOGNIZE, and FOR SYSTEM_TIME AS OF, did not receive
recent enough watermarks.

The current time of an operator is calculated by the minimum watermark of all
inputs, meaning across all tables/topics and their partitions.

If one partition does not emit a watermark, it can affect the entire pipeline.

The following statements can help with debugging issues related to
watermarks.

```mysql
-- Some example table
CREATE TABLE t_watermark_debugging (key INT, s STRING)
  DISTRIBUTED BY (key) INTO 4 BUCKETS
  WITH (
    'key.format' = 'raw',
    'value.format' = 'json-registry'
  );

-- Each value lands in a separate Kafka partition (out of 4).
-- Leave out values to see missing watermarks.
INSERT INTO t_watermark_debugging
  VALUES (9, 'Bob'), (10, 'Alice'), (8, 'John'), (2, 'David');

-- Idle partition detection is enabled by default. After at
-- most 10s to 5min, the partition is marked as idle and the maximum
-- timestamp is sent as a watermark.
-- Disable the idle timeout, to see missing watermarks for a partition.
SET 'sql.tables.scan.idle-timeout' = '0';

-- If ROW_NUMBER doesn't show results, it's clearly a watermark issue.
SELECT ROW_NUMBER() OVER (ORDER BY $rowtime ASC) AS `number`, *
  FROM t_watermark_debugging;

-- Add partition information as metadata column
ALTER TABLE t_watermark_debugging ADD part INT METADATA FROM 'partition' VIRTUAL;

-- Use the CURRENT_WATERMARK() function to check which watermark is calculated
SELECT
  *,
  part AS `Row Partition`,
  $rowtime AS `Row Timestamp`,
  CURRENT_WATERMARK($rowtime) AS `Operator Watermark`
FROM t_watermark_debugging;

-- Visualize the highest timestamp per Kafka partition
-- Due to the table declaration of 4 buckets, this query should show 4 rows.
-- If not, the missing partitions might be the cause for watermark issues.
SELECT part AS `Partition`, MAX($rowtime) AS `Max Timestamp in Partition`
  FROM t_watermark_debugging
  GROUP BY part;

-- A workaround could be to not use the system watermark:
ALTER TABLE t_watermark_debugging
  MODIFY WATERMARK FOR $rowtime AS $rowtime - INTERVAL '2' SECOND;
-- Or for perfect input data:
ALTER TABLE t_watermark_debugging
  MODIFY WATERMARK FOR $rowtime AS $rowtime - INTERVAL '0.001' SECOND;

-- Add "fresh" data while the above statements with
-- ROW_NUMBER() or CURRENT_WATERMARK() are running.
INSERT INTO t_watermark_debugging VALUES
  (9, 'Fresh Bob'),
  (10, 'Fresh Alice'),
  (8, 'Fresh John'),
  (2, 'Fresh David');
```

The preceding debugging examples don’t solve everything, but they can help
you find the root cause.

By default, when idleness detection is enabled, it forwards the latest event time from each
partition before marking it as idle, and then excludes idle partitions
from the watermark calculation. At least one partition must produce new
data for the watermarks to keep advancing.

Typically, root causes are:

- Idle Kafka partitions
- No data in Kafka partitions
- Watermark strategy is too conservative
- No fresh data after warm up with historical data for progressing the logical
  clock

### Handle idle partitions for missing watermarks

Idle partitions often cause missing watermarks. Also, no data in a partition or
infrequent data can be a root cause.

```mysql
-- Create a topic with 4 partitions.
CREATE TABLE t_watermark_idle (k INT, s STRING)
  DISTRIBUTED BY (k) INTO 4 BUCKETS;

-- Use a custom watermark with a coarser tolerance for this example.
ALTER TABLE t_watermark_idle
  MODIFY WATERMARK FOR $rowtime AS $rowtime - INTERVAL '2' SECONDS;

-- Each value lands in a separate Kafka partition, and partition 1 is empty.
INSERT INTO t_watermark_idle
  VALUES
    (1, 'Bob in partition 0'),
    (2, 'Alice in partition 3'),
    (8, 'John in partition 2');

-- Thread 1: Start a streaming job.
SELECT ROW_NUMBER() OVER (ORDER BY $rowtime ASC) AS `number`, *
  FROM t_watermark_idle;

-- Thread 2: Insert some data immediately -> Thread 1 still without results.
INSERT INTO t_watermark_idle
  VALUES (1, 'Another Bob in partition 0 shortly after');

-- Thread 2: Insert some data after 10s -> Thread 1 should show results.
INSERT INTO t_watermark_idle
  VALUES (1, 'Another Bob in partition 0 after 10s')
```

Within the first 10 seconds, all partitions contribute to the watermark
calculation, so the first INSERT INTO has no effect because partition 1 is
still empty.

After 10 seconds, all partitions are marked as idle. No partition contributes
to the watermark calculation. But when the second INSERT INTO is executed, it
becomes the main driving partition for the logical clock.

The global watermark jumps to “second INSERT INTO - 2 seconds”.

In the following code, the `sql.tables.scan.idle-timeout` configuration overrides
the default idle-detection algorithm, so even an immediate INSERT INTO can be
the main driving partition for the logical clock, because all other partitions
are marked as idle after 1 second.

```mysql
-- Thread 1: Start a streaming job.
-- Lower the idle timeout further.
SET 'sql.tables.scan.idle-timeout' = '1s';
SELECT ROW_NUMBER() OVER (ORDER BY $rowtime ASC) AS `number`, *
  FROM t_watermark_idle;

-- Thread 2: Insert some data immediately -> Thread 1 should show results.
INSERT INTO t_watermark_idle
  VALUES (1, 'Another Bob in partition 0 shortly after');
```

### Change the schema context property

You can set the schema context for key and value formats to control the namespace
for your schema resolution in Schema Registry.

1. Set the schema context for the value format
   ```sql
   ALTER TABLE `orders` SET ('value.format.schema-context' = '.lsrc-newcontext');
   ```

   Your output should resemble:
   ```none
   Statement phase is COMPLETED.
   ```
2. Check the new table properties.
   ```sql
   SHOW CREATE TABLE `orders`;
   ```

   Your output should resemble:
   ```text
   +----------------------------------------------------------------------+
   |                          SHOW CREATE TABLE                           |
   +----------------------------------------------------------------------+
   | CREATE TABLE `catalog`.`database`.`orders` (                         |
   |   `user` BIGINT NOT NULL,                                            |
   |   `product` VARCHAR(2147483647),                                     |
   |   `amount` INT,                                                      |
   |   `ts` TIMESTAMP(3)                                                  |
   | )                                                                    |
   |   DISTRIBUTED BY HASH(`user`) INTO 6 BUCKETS                         |
   | WITH (                                                               |
   |   'changelog.mode' = 'upsert',                                       |
   |   'connector' = 'confluent',                                         |
   |   'kafka.cleanup-policy' = 'delete',                                 |
   |   'kafka.max-message-size' = '2097164 bytes',                        |
   |   'kafka.retention.size' = '0 bytes',                                |
   |   'kafka.retention.time' = '604800000 ms',                           |
   |   'key.format' = 'avro-registry',                                    |
   |   'scan.bounded.mode' = 'unbounded',                                 |
   |   'scan.startup.mode' = 'latest-offset',                             |
   |   'value.format' = 'avro-registry',                                  |
   |   'value.format.schema-context' = '.lsrc-newcontext'                 |
   | )                                                                    |
   |                                                                      |
   +----------------------------------------------------------------------+
   ```

## Inferred tables schema evolution

You can use the ALTER TABLE statement to evolve schemas for
[inferred tables](#flink-sql-examples-inferred-tables).

The following examples show output from the SHOW CREATE TABLE statement called
on the resulting table.

### Schema Registry columns overlap with computed/metadata columns

For the following value schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "uid",
      "type": "int"
    }
  ]
}
```

Evolve a table by adding metadata:

```sql
ALTER TABLE t_metadata_overlap ADD `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA;
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE t_metadata_overlap` (
  `key` VARBINARY(2147483647),
  `uid` INT NOT NULL,
  `timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL METADATA
) DISTRIBUTED BY HASH(`key`) INTO 6 BUCKETS
WITH (
  ...
)
```

Properties
: - Schema Registry says there is a timestamp physical column, but Flink says there is
    timestamp metadata column.
  - In this case, metadata columns and computed columns have precedence, and
    Confluent Cloud for Apache Flink removes the physical column from the schema.
  - Because Confluent Cloud for Apache Flink advertises [FULL_TRANSITIVE mode](../../sr/fundamentals/schema-evolution.md#sr-compatibility-types),
    queries still work, and the physical column is set to NULL in the payload:
    ```sql
    INSERT INTO t_metadata_overlap
      SELECT CAST(NULL AS BYTES), 42, TO_TIMESTAMP_LTZ(0, 3);
    ```

Evolve the table by renaming metadata:

```sql
ALTER TABLE t_metadata_overlap DROP `timestamp`;

ALTER TABLE t_metadata_overlap
  ADD message_timestamp TIMESTAMP_LTZ(3) METADATA FROM 'timestamp';

SELECT * FROM t_metadata_overlap;
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_metadata_overlap` (
  `key` VARBINARY(2147483647),
  `uid` INT NOT NULL,
  `timestamp` VARCHAR(2147483647),
  `message_timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE METADATA FROM 'timestamp'
) DISTRIBUTED BY HASH(`key`) INTO 6 BUCKETS
WITH (
  ...
)
```

Properties
: - Now, both physical and metadata columns appear and can be accessed for
    reading and writing.

### Enrich a column that has no Schema Registry information

For the following value schema in Schema Registry:

```json
{
  "type": "record",
  "name": "TestRecord",
  "fields": [
    {
      "name": "uid",
      "type": "int"
    }
  ]
}
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_enrich_raw_key` (
  `key` VARBINARY(2147483647),
  `uid` INT NOT NULL
  ) DISTRIBUTED BY HASH(`key`) INTO 6 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'key.format' = 'raw',
  'value.format' = 'avro-registry'
  ...
)
```

Properties
: - Schema Registry provides only information for the value part.
  - Because the `key` part is not backed by Schema Registry, the `key.format` is
    `raw`.
  - The default data type of `raw` is BYTES, but you can change this by
    using the ALTER TABLE statement.

Evolve the table by giving a raw format column a specific type:

```sql
ALTER TABLE t_enrich_raw_key MODIFY key STRING;
```

SHOW CREATE TABLE returns the following output:

```sql
CREATE TABLE `t_enrich_raw_key` (
  `key` STRING,
  `uid` INT NOT NULL
) DISTRIBUTED BY HASH(`key`) INTO 6 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'key.format' = 'raw',
  'value.format' = 'avro-registry'
  ...
)
```

Properties
: - Only changes to simple, atomic types, such as INT, BYTES, and STRING, are
    supported, where the binary representation is clear.
  - For more complex modifications, use Schema Registry.
  - In multi-cluster scenarios, the ALTER TABLE statement must be executed for
    every cluster, because the data type for `key` is stored in the Flink
    regional metastore.

### Configure Schema Registry subject names

When working with topics that use RecordNameStrategy or TopicRecordNameStrategy,
you can configure the subject names for the schema resolution in Schema Registry. This is
particularly useful when handling multiple event types in a single topic.

For topics using these strategies, Flink initially infers a raw binary table:

```sql
SHOW CREATE TABLE events;
```

Your output shows a raw binary structure:

```sql
CREATE TABLE `events` (
  `key` VARBINARY(2147483647),
  `value` VARBINARY(2147483647)
) DISTRIBUTED BY HASH(`key`) INTO 6 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'key.format' = 'raw',
  'value.format' = 'raw'
)
```

Configure value schema subject names for each format:

### Avro

```sql
ALTER TABLE events SET (
  'value.format' = 'avro-registry',
  'value.avro-registry.subject-names' = 'com.example.Order;com.example.Shipment'
);
```

### JSON Schema

```sql
ALTER TABLE events SET (
  'value.format' = 'json-registry',
  'value.json-registry.subject-names' = 'com.example.Order;com.example.Shipment'
);
```

### Protobuf

```sql
ALTER TABLE events SET (
  'value.format' = 'proto-registry',
  'value.proto-registry.subject-names' = 'com.example.Order;com.example.Shipment'
);
```

If your topic uses keyed messages, you can also configure the key format:

```sql
ALTER TABLE events SET (
  'key.format' = 'avro-registry',
  'key.avro-registry.subject-names' = 'com.example.OrderKey'
);
```

You can configure both key and value schema subject names in a single statement:

```sql
ALTER TABLE events SET (
  'key.format' = 'avro-registry',
  'key.avro-registry.subject-names' = 'com.example.OrderKey',
  'value.format' = 'avro-registry',
  'value.avro-registry.subject-names' = 'com.example.Order;com.example.Shipment'
);
```

Properties:
: - Use semicolons (`;`) to separate multiple subject names
  - Subject names must match exactly with the names registered in Schema Registry
  - The format prefix (`avro-registry`, `json-registry`, or
    `proto-registry`) must match the schema format in Schema Registry

### Reset a key value

You can use the RESET option to set any key to its default value.

The following example shows how to reset a table that has a JSON Schema
back to raw format.

```sql
ALTER TABLE json_table RESET (
  'value.json-registry.wire-encoding',
  'value.json-registry.subject-names'
);
```

## Custom error handling

You can use ALTER TABLE with the
[error-handling.mode](statements/create-table.md#flink-sql-create-table-with-error-handling-mode)
and
[error-handling.log.target](statements/create-table.md#flink-sql-create-table-with-error-handling-log-target)
table properties to set custom error handling for deserialization errors.

The following code example shows how to log errors to the specified Dead Letter
Queue (DLQ) table and enable processing to continue.

```sql
 ALTER TABLE my_table SET (
  'error-handling.mode' = 'log',
  'error-handling.log.target' = 'my_error_table'
);
```

#### NOTE
Custom error handling applies only to deserialization errors at the source.
Confluent Cloud for Apache Flink does not route errors that occur inside
[user-defined functions (UDFs)](../concepts/user-defined-functions.md#flink-sql-udfs) to the DLQ. If an unhandled exception occurs in a UDF, the
statement fails. For recommendations on handling errors inside UDFs, see
[Error handling best practices](../how-to-guides/create-udf.md#flink-sql-udf-error-handling-best-practices).

For detailed DLQ configuration options, including how to pre-create a DLQ with
custom topic settings and the required schemas, see
[Configure a Dead Letter Queue](../how-to-guides/configure-dlq.md#flink-sql-configure-dlq).

<!-- Query errors by using the ``$errors`` system view: -->
<!-- .. code:: mysql -->
<!-- SELECT * FROM my_table$errors; -->

## Late data handling

Configure how the source operator handles late-arriving events that arrive after
the watermark has advanced past their event timestamp.

### Filter late data to System Tables

Enable filtering to separate on-time data from late data. Late events are
preserved in a System Table (`Orders$late`) for inspection or reprocessing:

```sql
ALTER TABLE Orders SET (
  'late-handling.mode' = 'filter'
);
```

### Create table with filtering enabled

Configure late data handling at table creation:

```sql
CREATE TABLE Orders (
  order_id INT,
  amount DECIMAL(10,2)
) WITH (
  'late-handling.mode' = 'filter'
);
```

### Revert to default behavior

Reset to the default `pass-through` mode:

```sql
ALTER TABLE Orders RESET ('late-handling.mode');
```

#### NOTE
- The default mode is `pass-through`, where late events flow to downstream
  operators. Window aggregations and other time-based operators handle late
  data according to their own policies.
- System Tables (`<table_name>$late`) are virtual views on the source topic
  and do not create additional physical storage.
- Late data availability in System Tables is governed by the source topic’s
  retention policy. To preserve late data longer, explicitly persist it with
  `INSERT INTO` or create a Materialized Table.
- Late data metrics (records per minute, max lateness) are reported regardless
  of the configured mode and are visible in the Flink UI and Metrics API.

For detailed configuration steps, see [Handle Late-Arriving Data](../how-to-guides/handle-late-arriving-data.md#handle-late-arriving-data).

**Properties**

| Property             | Description                                                                                                                                                                                                                               |
|----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `late-handling.mode` | Controls how the source handles late events. Values: `pass-through`<br/>(default) or `filter`. When set to `filter`, late events are<br/>separated from the main stream and made available in a System Table<br/>with the `$late` suffix. |

## Kafka topic configuration

You can use ALTER TABLE to modify Kafka topic-level configuration options for
your tables.

### Change cleanup policy

Modify the cleanup policy for the underlying Kafka topic. For more information,
see [kafka.cleanup-policy](statements/create-table.md#flink-sql-create-table-with-kafka-cleanup-policy).

```sql
-- Change to delete policy (old segments discarded after retention period)
ALTER TABLE my_table SET (
  'kafka.cleanup-policy' = 'delete'
);

-- Change to compact policy (log compaction enabled)
ALTER TABLE my_table SET (
  'kafka.cleanup-policy' = 'compact'
);

-- Change to delete-compact policy (both compaction and deletion)
ALTER TABLE my_table SET (
  'kafka.cleanup-policy' = 'delete-compact'
);
```

### Modify retention settings

Configure retention time and size limits for the underlying Kafka topic. For
more information, see [kafka.retention.time](statements/create-table.md#flink-sql-create-table-with-kafka-retention-time)
and [kafka.retention.size](statements/create-table.md#flink-sql-create-table-with-kafka-retention-size).

```sql
-- Set retention time to 30 days
ALTER TABLE my_table SET (
  'kafka.retention.time' = '30 days'
);

-- Set retention size to 100 GB
ALTER TABLE my_table SET (
  'kafka.retention.size' = '100 GB'
);

-- Set both retention time and size
ALTER TABLE my_table SET (
  'kafka.retention.time' = '7 days',
  'kafka.retention.size' = '50 GB'
);
```

### Configure compaction time

Set the minimum time a message remains uncompacted in the log for upsert
tables. For more information, see
[kafka.compaction.time](statements/create-table.md#flink-sql-create-table-with-kafka-compaction-time).

```sql
-- Set compaction delay to 3 days
ALTER TABLE my_upsert_table SET (
  'kafka.compaction.time' = '3 days'
);
```

### Adjust maximum message size

Change the maximum message size for the underlying Kafka topic. For more
information, see [kafka.max-message-size](statements/create-table.md#flink-sql-create-table-with-kafka-max-message-size).

```sql
-- Increase max message size to 5 MB
ALTER TABLE my_table SET (
  'kafka.max-message-size' = '5 MB'
);
```

### Configure producer compression

Set the compression type for messages produced to the underlying Kafka topic.
For more information, see
[kafka.producer.compression.type](statements/create-table.md#flink-sql-create-table-with-kafka-producer-compression-type).

```sql
-- Enable gzip compression
ALTER TABLE my_table SET (
  'kafka.producer.compression.type' = 'gzip'
);

-- Enable zstd compression (recommended for better compression ratio)
ALTER TABLE my_table SET (
  'kafka.producer.compression.type' = 'zstd'
);

-- Disable compression
ALTER TABLE my_table SET (
  'kafka.producer.compression.type' = 'none'
);
```

### Set consumer isolation level

Control which transactional messages to read from the underlying Kafka topic.
For more information, see
[kafka.consumer.isolation-level](statements/create-table.md#flink-sql-create-table-with-kafka-consumer-isolation-level).

```sql
-- Read only committed messages (default, recommended for most use cases)
ALTER TABLE my_table SET (
  'kafka.consumer.isolation-level' = 'read-committed'
);

-- Read all messages including uncommitted
ALTER TABLE my_table SET (
  'kafka.consumer.isolation-level' = 'read-uncommitted'
);
```

## Scan configuration

You can use ALTER TABLE to modify how tables are scanned when reading data.

### Change scan startup mode

Modify where the consumer starts reading from the underlying Kafka topic. For
more information, see [scan.startup.mode](statements/create-table.md#flink-sql-create-table-with-scan-startup-mode).

```sql
-- Start reading from the earliest offset
ALTER TABLE my_table SET (
  'scan.startup.mode' = 'earliest-offset'
);

-- Start reading from the latest offset
ALTER TABLE my_table SET (
  'scan.startup.mode' = 'latest-offset'
);

-- Start reading from a specific timestamp
ALTER TABLE my_table SET (
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1609459200000'
);
```

### Configure bounded mode

Set bounded mode to control where the consumer stops reading. For more
information, see [scan.bounded.mode](statements/create-table.md#flink-sql-create-table-with-scan-bounded-mode).

```sql
-- Read until latest offset (bounded read)
ALTER TABLE my_table SET (
  'scan.bounded.mode' = 'latest-offset'
);

-- Read until specific timestamp (bounded read)
ALTER TABLE my_table SET (
  'scan.bounded.mode' = 'timestamp',
  'scan.bounded.timestamp-millis' = '1640995200000'
);

-- Return to unbounded mode (continuous streaming)
ALTER TABLE my_table SET (
  'scan.bounded.mode' = 'unbounded'
);
```

## Related content

- Video: [How to Set Idle Timeouts](https://www.youtube.com/watch?v=YSIhM5-Sykw)
- [CREATE TABLE WITH options](statements/create-table.md#flink-sql-with-options)

<a id="flink-sql-examples-select"></a>

## SELECT examples

The following examples show frequently used scenarios for SELECT.

### Most minimal statement

The shortest valid SELECT statement returns a single literal value and runs
as a bounded query.

Syntax
: ```sql
  SELECT 1;
  ```

Properties
: - Statement is bounded

### Verify the local time zone configuration

Syntax
: ```sql
  SELECT NOW();
  ```

Properties
: - Statement is bounded
  - NOW() returns a TIMESTAMP_LTZ(3), so if the client is configured correctly,
    it should show a timestamp in your local time zone.

### Combine multiple tables into one

Syntax
: ```sql
  CREATE TABLE t_union_1 (i INT);
  CREATE TABLE t_union_2 (i INT);
  TABLE t_union_1 UNION ALL TABLE t_union_2;
  <br/>
  -- alternate syntax
  SELECT * FROM t_union_1
  UNION ALL
  SELECT * FROM t_union_2;
  ```

### Get insights into the current watermark

Syntax
: ```mysql
  CREATE TABLE t_watermarked_insight (s STRING) DISTRIBUTED INTO 1 BUCKETS;
  <br/>
  INSERT INTO t_watermarked_insight VALUES ('Bob'), ('Alice'), ('Charly');
  <br/>
  SELECT $rowtime, CURRENT_WATERMARK($rowtime) FROM t_watermarked_insight;
  ```
  <br/>
  The output resembles:
  <br/>
  ```none
  $rowtime                EXPR$1
  2024-04-29 11:59:01.080 NULL
  2024-04-29 11:59:01.093 2024-04-04 15:27:37.433
  2024-04-29 11:59:01.094 2024-04-04 15:27:37.433
  ```

Properties
: - The CURRENT_WATERMARK function returns the watermark that arrived at the
    operator evaluating the SELECT statement.
  - The returned watermark is the minimum of all inputs, across all
    tables/topics and their partitions.
  - If a common watermark was not received from all inputs, the function
    returns NULL.
  - The CURRENT_WATERMARK function takes a
    [time attribute](../concepts/timely-stream-processing.md#flink-sql-time-attributes), which is a column
    that has WATERMARK FOR defined.

A watermark is always emitted after the row has been processed, so the first
row always has a NULL watermark.

The default watermark strategy uses a fixed out-of-orderness tolerance of
180 ms with no warmup period. The watermark advances immediately as data
arrives, trailing the maximum observed event time by 180 ms.

Sources emit watermarks every 200 ms, but within the first 200 ms they emit
per row for powering examples such as this one.

### Flatten fields into columns

Syntax
: ```sql
  CREATE TABLE t_flattening (i INT, r1 ROW<i INT, s STRING>, r2 ROW<other INT>);
  <br/>
  SELECT r1.*, r2.* FROM t_flattening;
  ```

Properties
: You can apply the `*` operator on nested data, which enables flattening
  fields into columns of the table.

<a id="flink-sql-examples-schema-reference"></a>

## Schema reference examples

The following examples show how to use schema references in Flink SQL.

<!-- flink-sql-examples-schema-reference_start -->

For the following schemas in Schema Registry:

### Avro

```json
{
   "type":"record",
   "namespace": "io.confluent.developer.avro",
   "name":"Purchase",
   "fields": [
      {"name": "item", "type":"string"},
      {"name": "amount", "type": "double"},
      {"name": "customer_id", "type": "string"}
   ]
}
```

### Protobuf

```proto
syntax = "proto3";

package io.confluent.developer.proto;

message Purchase {
   string item = 1;
   double amount = 2;
   string customer_id = 3;
}
```

### JSON

```json
{
   "$schema": "http://json-schema.org/draft-07/schema#",
   "title": "Purchase",
   "type": "object",
   "properties": {
      "item": {
         "type": "string"
      },
      "amount": {
         "type": "number"
      },
      "customer_id": {
         "type": "string"
      }
   },
   "required": ["item", "amount", "customer_id"]
}
```

<br/>

### Avro

```json
{
   "type":"record",
   "namespace": "io.confluent.developer.avro",
   "name":"Pageview",
   "fields": [
      {"name": "url", "type":"string"},
      {"name": "is_special", "type": "boolean"},
      {"name": "customer_id", "type":  "string"}
   ]
}
```

### Protobuf

```proto
syntax = "proto3";

package io.confluent.developer.proto;

message Pageview {
   string url = 1;
   bool is_special = 2;
   string customer_id = 3;
}
```

### JSON

```json
{
   "$schema": "http://json-schema.org/draft-07/schema#",
   "title": "Pageview",
   "type": "object",
   "properties": {
      "url": {
         "type": "string"
      },
      "is_special": {
         "type": "boolean"
      },
      "customer_id": {
         "type": "string"
      }
   },
   "required": ["url", "is_special", "customer_id"]
}
```

<br/>

### Avro

```json
[
   "io.confluent.developer.avro.Purchase",
   "io.confluent.developer.avro.Pageview"
]
```

### Protobuf

```proto
syntax = "proto3";

package io.confluent.developer.proto;

import "purchase.proto";
import "pageview.proto";

message CustomerEvent {
   oneof action {
      Purchase purchase = 1;
      Pageview pageview = 2;
   }
}
```

### JSON

```json
{
   "$schema": "http://json-schema.org/draft-07/schema#",
   "title": "CustomerEvent",
   "type": "object",
   "oneOf": [
      { "$ref": "io.confluent.developer.json.Purchase" },
      { "$ref": "io.confluent.developer.json.Pageview" }
   ]
}
```

<br/>

and references:

### Avro

```json
[
   {
      "name": "io.confluent.developer.avro.Purchase",
      "subject": "purchase",
      "version": 1
   },
   {
      "name": "io.confluent.developer.avro.Pageview",
      "subject": "pageview",
      "version": 1
   }
]
```

### Protobuf

```json
[
   {
      "name": "purchase.proto",
      "subject": "purchase",
      "version": 1
   },
   {
      "name": "pageview.proto",
      "subject": "pageview",
      "version": 1
   }
]
```

### JSON

```json
[
   {
      "name": "io.confluent.developer.json.Purchase",
      "subject": "purchase",
      "version": 1
   },
   {
      "name": "io.confluent.developer.json.Pageview",
      "subject": "pageview",
      "version": 1
   }
]
```

<br/>

`SHOW CREATE TABLE customer-events;` returns the following output:

```sql
CREATE TABLE `customer-events` (
  `key` VARBINARY(2147483647),
  `Purchase` ROW<`item` VARCHAR(2147483647) NOT NULL, `amount` DOUBLE NOT NULL, `customer_id` VARCHAR(2147483647) NOT NULL>,
  `Pageview` ROW<`url` VARCHAR(2147483647) NOT NULL, `is_special` BOOLEAN NOT NULL, `customer_id` VARCHAR(2147483647) NOT NULL>
)
DISTRIBUTED BY HASH(`key`) INTO 2 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'connector' = 'confluent',
  'kafka.cleanup-policy' = 'delete',
  'kafka.max-message-size' = '2097164 bytes',
  'kafka.retention.size' = '0 bytes',
  'kafka.retention.time' = '7 d',
  'key.format' = 'raw',
  'scan.bounded.mode' = 'unbounded',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = '[VALUE_FORMAT]'
)
```

### Split into tables for each type

**Syntax**

```sql
CREATE TABLE purchase AS
   SELECT Purchase.* FROM `customer-events`
   WHERE Purchase IS NOT NULL;

SELECT * FROM purchase;
```

```sql
CREATE TABLE pageview AS
   SELECT Pageview.* FROM `customer-events`
   WHERE Pageview IS NOT NULL;

SELECT * FROM pageview;
```

Output:

| item   |   amount | customer_id   |
|--------|----------|---------------|
| apple  |     9.99 | u-21          |
| jam    |     4.29 | u-67          |
| mango  |    13.99 | u-67          |
| socks  |     7.99 | u-123         |

| url                                                  | is_special   | customer_id   |
|------------------------------------------------------|--------------|---------------|
| [https://www.confluent.io](https://www.confluent.io) | TRUE         | u-67          |
| [http://www.cflt.io](http://www.cflt.io)             | FALSE        | u-12          |
<!-- flink-sql-examples-schema-reference_end -->

## Related content

- [CREATE TABLE](statements/create-table.md#flink-sql-create-table)
- [ALTER TABLE](statements/alter-table.md#flink-sql-alter-table)
- [Flink SQL Queries](queries/overview.md#flink-sql-queries)
- [Flink SQL Functions](functions/overview.md#flink-sql-functions-overview)
- [DDL Statements in Confluent Cloud for Apache Flink](statements/overview.md#flink-sql-statements-overview)
- [Data Type Mappings](serialization.md#flink-sql-serialization)
- [Schema Registry](../../sr/schemas-manage.md#sr-prv)
- [SHOW Statements](statements/show.md#flink-sql-show)

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