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.

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.

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<INT>` column to `ARRAY<BIGINT>`, or a `MAP<STRING, FLOAT>` column to `MAP<STRING, DOUBLE>`.	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.

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<STRING>` column to `ARRAY<STRING NOT NULL>`.
Change the key type of a `MAP`	Change a `MAP<INT, STRING>` column to `MAP<BIGINT, STRING>`. You can widen only the value type of a `MAP`, not the key type.
Change the key schema of a topic in 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.

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:

{
  "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:

{
  "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:

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.

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:

Create a new Kafka topic. For the steps to create a topic, see Create topics.
Register the new schema for the new topic in Schema Registry. For the steps to register a schema, see Create a schema.
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.

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.
Enable a Lightning Table on the new topic. For the steps to enable a Lightning Table, see Enable a Lightning Table for a topic.
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.
Update any applications and agents that query the original topic so that they query the new topic.
Disable the Lightning Table on the original topic. For the steps to disable a Lightning Table, see Disable a Lightning Table.