# Schema Evolution for Lightning Tables in Confluent Cloud A Lightning Table uses the schema of its Apache Kafka® topic in Schema Registry. When you evolve the topic’s schema, the Lightning Table checks whether it supports the change. This check is separate from the checks that Schema Registry runs. Schema Registry uses a compatibility mode to decide which new schema versions it accepts for a subject. Even if Schema Registry accepts a change, a Lightning Table might not support it. For example, the `NONE` compatibility mode accepts a renamed column, but a Lightning Table doesn’t support renamed columns. For more information about compatibility modes, see [Schema Evolution and Compatibility for Schema Registry on Confluent Cloud](../sr/fundamentals/schema-evolution.md#schema-evolution-and-compatibility). If the Lightning Table supports the change, it keeps serving queries over your old and new records together. If it doesn’t support the change, it stops ingesting new records. To fix the Lightning Table, revert the change or move to a new topic. For the steps, see [Recover from an unsupported schema change](#lightning-tables-schema-evolution-recover). ## How a Lightning Table handles schema changes After a supported schema change, a Lightning Table does the following: - Keeps the records that it already stores, without rewriting them. - Serves records written before and after the change together, as one table. - Returns `NULL` for a new column or field in records that were written before you added it. - Returns the values of a widened column as the wider type, including values written before the change. - Takes some time to reflect the new schema in query results. The same rules apply to topics that use Avro, Protobuf, or JSON Schema. ## Supported schema changes The following table lists the schema changes that a Lightning Table supports. For each change, it gives an example. It also describes the query results for records that were written before you changed the schema. The table uses SQL type names, which are the same for every schema format. | Change | Example | Records written before the change | |-----------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------| | Add a nullable column | Add an optional `discount_code` column of type `STRING`. | Return `NULL` for the new column. | | Add a nullable field to a `ROW` | Add an optional field to a `ROW` column, or to a `ROW` nested
at any depth inside an `ARRAY`, a `MAP`, or another `ROW`. | Return `NULL` for the new field. | | Widen an integer type | Change an `INT` column to `BIGINT`. | Return values as `BIGINT`. | | Widen a floating-point type | Change a `FLOAT` column to `DOUBLE`. | Return values as `DOUBLE`. | | Increase `DECIMAL` precision | Change a `DECIMAL(10, 2)` column to `DECIMAL(15, 2)`. | Return values with the new precision. | | Widen the element type of an `ARRAY` or the value type of a
`MAP` | Change an `ARRAY` column to `ARRAY`, or a
`MAP` column to `MAP`. | Return elements or values as the wider type. | | Make a required column nullable | Change a `NOT NULL` column to allow `NULL`. | Return the same values as before. | ## Unsupported schema changes If you make an unsupported schema change, the Lightning Table stops ingesting new records, and queries on it return an error. The Lightning Table’s status stays **Active**. You must fix the Lightning Table to resume ingestion. For the steps to fix it, see [Recover from an unsupported schema change](#lightning-tables-schema-evolution-recover). The following table lists the schema changes that a Lightning Table doesn’t support, with an example of each. | Change | Example | |----------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------| | Drop a column | Remove the `discount_code` column. | | Rename a column | Rename the `quantity` column to `item_count`. | | Reorder columns | Move the `quantity` column before the `order_id` column. | | Add a required column | Add a `region` column that’s `NOT NULL`. | | Narrow a type | Change a `BIGINT` column to `INT`. | | Change a type incompatibly | Change a `STRING` column to `INT`. | | Change `DECIMAL` scale | Change a `DECIMAL(10, 2)` column to `DECIMAL(10, 4)`. | | Make a nullable column required | Change a column that allows `NULL` to `NOT NULL`. | | Make the element type of an `ARRAY` or the value type of a
`MAP` required | Change an `ARRAY` column to `ARRAY`. | | Change the key type of a `MAP` | Change a `MAP` column to `MAP`.
You can widen only the value type of a `MAP`, not the key type. | | Change the key schema of a topic in
[upsert mode](limitations.md#rtce-upsert-mode) | Add a `region` field to the key schema of a compacted topic
whose record key has a schema. | The rules in this table also apply to fields inside a `ROW`, including a `ROW` inside an `ARRAY` or `MAP`. For example, you can’t drop, rename, or reorder a nested field. ## Example: Add a column and widen a type To evolve the schema for a Lightning Table, register a new schema version for its topic in Schema Registry. For the steps to register a schema version, see [Edit schemas](../sr/schemas-manage.md#cloud-schema-edit). The following example adds a nullable column and widens a type for an `orders` topic that uses Avro. The same changes work for Protobuf and JSON Schema. The topic starts with the following schema: ```json { "type": "record", "name": "Order", "fields": [ {"name": "order_id", "type": "string"}, {"name": "quantity", "type": "int"} ] } ``` The topic has two records that use this schema: | order_id | quantity | |------------|------------| | ORD-1 | 2 | | ORD-2 | 5 | The new schema version widens `quantity` from `int` to `long` and adds a nullable `discount_code` field: ```json { "type": "record", "name": "Order", "fields": [ {"name": "order_id", "type": "string"}, {"name": "quantity", "type": "long"}, {"name": "discount_code", "type": ["null", "string"], "default": null} ] } ``` After you register the new schema version, a producer sends a message. The message has an `order_id` of `ORD-3`, a `quantity` of `3000000000`, and a `discount_code` of `FALL10`. After the Lightning Table ingests the new record, you run the following query on it: ```sql SELECT order_id, quantity, discount_code FROM orders ORDER BY order_id ``` The query returns both the old and new records: | order_id | quantity | discount_code | |------------|------------|-----------------| | ORD-1 | 2 | NULL | | ORD-2 | 5 | NULL | | ORD-3 | 3000000000 | FALL10 | The `quantity` column is now `BIGINT` for every record. The old records return `NULL` for `discount_code`. ## Recover from an unsupported schema change To fix a Lightning Table after an unsupported schema change, do one of the following: - **Revert the change**: If you don’t need the new schema, register the previous schema again in Schema Registry. The Lightning Table resumes ingesting new records automatically, even if producers already wrote records with the new schema. - **Move to a new topic**: If you need the new schema, create a new topic that uses it, and enable a Lightning Table on the new topic. To bring your existing data along, follow the steps in [Copy data to a new topic](#lightning-tables-schema-evolution-copy-data). You can’t fix the Lightning Table by disabling it and enabling it again. ### Copy data to a new topic Use Apache Flink® to copy your data from the original topic to a new topic that uses the new schema. The original topic’s retention must cover all the data that you want to copy. To copy your data to a new topic, complete the following steps: 1. Create a new Kafka topic. For the steps to create a topic, see [Create topics](../topics/overview.md#create-topics-cloud). 2. Register the new schema for the new topic in Schema Registry. For the steps to register a schema, see [Create a schema](../sr/schemas-manage.md#cloud-schema-create). 3. Start a continuous Flink SQL statement that reads from the original topic and writes to the new topic. For the syntax of a statement that copies data between topics, see [INSERT INTO FROM SELECT Statement in Confluent Cloud for Apache Flink](../flink/reference/queries/insert-into-from-select.md#flink-sql-insert-into-from-select-statement). The statement applies any transformations that the new schema needs, such as renaming a column, casting a type, or removing a column. Start the statement before you enable the Lightning Table, so that data is already flowing. 4. Enable a Lightning Table on the new topic. For the steps to enable a Lightning Table, see [Enable a Lightning Table for a topic](manage.md#lightning-tables-manage-enable). 5. Verify that the new Lightning Table returns the expected data. For the steps to query a Lightning Table, see [Query a Lightning Table in the Cloud Console](manage.md#lightning-tables-manage-query). 6. Update any applications and agents that query the original topic so that they query the new topic. 7. Disable the Lightning Table on the original topic. For the steps to disable a Lightning Table, see [Disable a Lightning Table](manage.md#lightning-tables-manage-disable). ## Related content - [Manage Lightning Tables in Confluent Cloud](manage.md#rtce-manage-topics) - [SQL Reference for Lightning Table Queries in Confluent Cloud](query/sql-reference.md#lightning-tables-sql-reference) - [Lightning Tables Support and Limitations in Confluent Cloud](limitations.md#rtce-limitations) - [Schema Evolution and Compatibility for Schema Registry on Confluent Cloud](../sr/fundamentals/schema-evolution.md#schema-evolution-and-compatibility) #### 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).