# Changelog Conversion `FROM_CHANGELOG` and `TO_CHANGELOG` give you explicit control over how Flink SQL interprets and produces changelog streams, converting between an external, custom changelog format and a Flink table. They’re built-in [process table functions (PTFs)](../../concepts/process-table-functions.md#flink-ptfs), so they’re ready to use directly in Flink SQL with no custom code to write or deploy. For an overview of PTFs in general, and how to write your own, see [Process Table Functions](../../concepts/process-table-functions.md#flink-ptfs). - [FROM_CHANGELOG](#flink-ptfs-from-changelog): reads an append-only stream that carries an operation code and turns it into an updating table. - [TO_CHANGELOG](#flink-ptfs-to-changelog): turns an updating table into an append-only stream of change events. For step-by-step examples with sample input and output, see [Read and write custom changelog formats](../../how-to-guides/read-write-custom-changelog.md#flink-read-write-changelog). #### NOTE In these function names, *changelog* refers to the external append-only stream that encodes each change with an explicit operation code. `FROM_CHANGELOG` reads that stream into a Flink table. `TO_CHANGELOG` writes a Flink table back out as that stream. ## FROM_CHANGELOG Reads an append-only stream that carries an operation code and turns it into an updating table that you can query, join, and aggregate over. Use it to interpret custom change data capture (CDC) or changelog formats that Flink doesn’t recognize natively. ### Syntax ```sql SELECT * FROM FROM_CHANGELOG( input => TABLE input_table [ PARTITION BY key ], op => DESCRIPTOR(op_column), op_mapping => MAP[ 'input_code', 'ROW_KIND', ... ], error_handling => 'FAIL' | 'SKIP' -- optional, defaults to FAIL ) ``` ### Parameters `input` : The append-only input table. To Flink, every record is an insert. Use `PARTITION BY` to route all events for a key to the same task. `op` : Names the field that holds the operation code. The column must be of type `STRING`. Defaults to `op`. `op_mapping` : Maps each operation code in your data to a Flink row kind: `INSERT`, `UPDATE_BEFORE`, `UPDATE_AFTER`, or `DELETE`. If you omit it, the operation codes must already match the row-kind names. `error_handling` : Optional. Controls what happens when an input row’s operation code is `NULL` or isn’t present in `op_mapping`. `FAIL`, the default, throws a runtime exception. `SKIP` silently drops the row and continues. ### Returns An updating table whose row kinds come from your `op_mapping`. If the mapping emits both `UPDATE_BEFORE` and `UPDATE_AFTER`, the result is a retract stream. If it emits `UPDATE_AFTER` without `UPDATE_BEFORE`, the result is an upsert stream keyed on the `PARTITION BY` columns, which must match the unique key of your data. #### IMPORTANT Known limitation: `FROM_CHANGELOG`’s upsert output (an `op_mapping` with no `UPDATE_BEFORE`) can’t be consumed directly by a foreground `SELECT` query. Instead, use a plain `INSERT INTO` or `CREATE TABLE ... AS SELECT`. A query built directly on that output, foreground or background, can fail to plan with an error like: ```none The query cannot be planned because of a changelog mode mismatch: an operator cannot produce the changelog its consumer requires. ``` This always happens for a foreground query. For a background query, it means something in the query is breaking upsert mode. Always materialize the upsert result into its own table first before running downstream queries. Querying `FROM_CHANGELOG(...)` directly with joins, aggregations, or filters can lead to silent data corruption—such as returning wrong results when a `DELETE` row contains only `NULL` values. In contrast, `TO_CHANGELOG` is unaffected because its output is strictly append-only. ### Default op_mapping If you omit `op_mapping`, the operation codes in your data must already be the Flink row-kind names, so a stream produced by `TO_CHANGELOG` reads back without extra configuration. | Input code | Row kind | |-----------------|---------------| | `INSERT` | INSERT | | `UPDATE_BEFORE` | UPDATE_BEFORE | | `UPDATE_AFTER` | UPDATE_AFTER | | `DELETE` | DELETE | You can map several input codes to one row kind by listing them in a single key, separated by commas, for example `'c, r'` for `INSERT`. Each row kind can appear at most once across the mapping. ### Output schema The output includes every input column except the operation-code column, which Flink interprets and removes. With `PARTITION BY`, the partition-key columns move to the front of the output. ### Examples #### Basic usage Read a stream that already uses the Flink row-kind names, with the default mapping: ```sql -- Input (append-only): -- +I[id:1, op:'INSERT', name:'Alice'] -- +I[id:2, op:'INSERT', name:'Bob'] -- +I[id:1, op:'UPDATE_BEFORE', name:'Alice'] -- +I[id:1, op:'UPDATE_AFTER', name:'Alice2'] -- +I[id:2, op:'DELETE', name:'Bob'] SELECT * FROM FROM_CHANGELOG(input => TABLE cdc_stream); -- Output (updating table): -- +I[id:1, name:'Alice'] -- +I[id:2, name:'Bob'] -- -U[id:1, name:'Alice'] -- +U[id:1, name:'Alice2'] -- -D[id:2, name:'Bob'] ``` After all events, the table holds one row: `id` 1 with `name` `'Alice2'`. #### Custom operation codes Map your own codes to row kinds. Here `c` is a create, `ub` and `ua` are the before and after of an update, `d` is a delete, and the codes are read from a column named `operation`: ```sql SELECT * FROM FROM_CHANGELOG( input => TABLE cdc_stream, op => DESCRIPTOR(operation), op_mapping => MAP[ 'c', 'INSERT', 'ub', 'UPDATE_BEFORE', 'ua', 'UPDATE_AFTER', 'd', 'DELETE' ] ); ``` #### Partition by a key Add `PARTITION BY` to route all events for a key to the same task. `FROM_CHANGELOG` assumes events for the same key arrive in order. `PARTITION BY` also moves the key columns to the front of the output, so prefer leaving it off unless a downstream operator is keyed on that column: ```sql -- Input columns (name, id, op, doc) become output columns (id, name, doc) SELECT * FROM FROM_CHANGELOG(input => TABLE cdc_stream PARTITION BY id); ``` #### Produce an upsert table To produce an upsert table, partition by the unique key and use an `op_mapping` that emits `UPDATE_AFTER` but not `UPDATE_BEFORE`. Flink then treats the partition key as the upsert key. Because upsert output can’t be consumed by a bare foreground `SELECT` (see the known limitation above), materialize it with `INSERT INTO` into a target table declared with a `PRIMARY KEY` and `'changelog.mode' = 'upsert'`: ```sql CREATE TABLE customers_upsert ( id INT, name STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'changelog.mode' = 'upsert' ); -- Upsert input: INSERT / UPDATE_AFTER / DELETE only INSERT INTO customers_upsert SELECT * FROM FROM_CHANGELOG( input => TABLE cdc_stream PARTITION BY id, op_mapping => MAP[ 'INSERT', 'INSERT', 'UPDATE_AFTER', 'UPDATE_AFTER', 'DELETE', 'DELETE' ] ); -- customers_upsert holds an upsert changelog, upsert key = id: -- +I[id:1, name:'Alice'] -- +I[id:2, name:'Bob'] -- +U[id:1, name:'Alice2'] -- -D[id:2, name:'Bob'] ``` #### Handle invalid operation codes By default, a row whose operation code is `NULL` or not in `op_mapping` fails the statement. Set `error_handling` to `SKIP` to drop those rows and continue instead: ```sql SELECT * FROM FROM_CHANGELOG( input => TABLE cdc_stream, error_handling => 'SKIP' ); ``` ## TO_CHANGELOG Turns an updating table into an append-only stream of change events, each carrying an explicit operation code. This is the first Flink SQL operator that can turn an upsert or retract stream back into an append stream. Use it to feed a downstream consumer that doesn’t understand the Flink internal changelog (and would otherwise read the `-U` and `+U` records as duplicates) or a compacted topic that expects CDC. ### Syntax ```sql SELECT * FROM TO_CHANGELOG( input => TABLE input_table [ PARTITION BY key ], op => DESCRIPTOR(op_column), op_mapping => MAP[ 'ROW_KIND', 'output_code', ... ], produces_full_deletes => TRUE -- optional, defaults to TRUE ) ``` ### Parameters `input` : The updating input table to convert. Accepts insert-only, retract, and upsert tables. For an upsert input, the `PARTITION BY` key should match or be a subset of the table’s upsert key. `op` : Names the operation-code column in the output. Defaults to `op`. `op_mapping` : Maps each Flink row kind to the operation code you want in the output. `produces_full_deletes` : Optional `BOOLEAN`. When `TRUE`, the default, each delete carries all columns, the full row image. When `FALSE`, only the key columns are kept and the rest are set to `null`. ### Returns An append-only table with the operation-code column added. Every output row is an `INSERT`, whatever the input row’s original change operation was. ### Default op_mapping If you omit `op_mapping`, each row kind maps to its standard name (`INSERT`, `UPDATE_BEFORE`, `UPDATE_AFTER`, `DELETE`), so `FROM_CHANGELOG` can read the result back without extra configuration. Provide an `op_mapping` to use your own codes. When you do, only the row kinds you map are forwarded and the rest are dropped, which is a simple way to filter. List several row kinds in one key, separated by commas, to map them to the same code. ### Output schema The output is the operation-code column followed by every input column. With `PARTITION BY`, the partition-key columns move to the front. ### Examples #### Basic usage Convert the upsert result of an aggregation into an append stream that labels each change: ```sql -- Input (upsert table from an aggregation): -- +I[name:'Alice', cnt:1] -- +U[name:'Alice', cnt:2] -- -D[name:'Bob', cnt:1] SELECT * FROM TO_CHANGELOG(input => TABLE my_aggregation); -- Output (append-only): -- +I[op:'INSERT', name:'Alice', cnt:1] -- +I[op:'UPDATE_AFTER', name:'Alice', cnt:2] -- +I[op:'DELETE', name:'Bob', cnt:1] ``` #### Custom operation codes and filtering Name the output column and map row kinds to your own codes. Any row kind you leave out of the mapping is dropped, so this also filters, for example to forward only inserts and updates: ```sql SELECT * FROM TO_CHANGELOG( input => TABLE my_aggregation, op => DESCRIPTOR(op_code), op_mapping => MAP[ 'INSERT', 'I', 'UPDATE_AFTER', 'U' ] ); -- Only INSERT and UPDATE_AFTER are forwarded; DELETE is dropped. ``` #### Upsert stream Map both `INSERT` and `UPDATE_AFTER` to one upsert code and `DELETE` to another, dropping `UPDATE_BEFORE`: ```sql SELECT * FROM TO_CHANGELOG( input => TABLE upsert_source PARTITION BY id, op_mapping => MAP[ 'INSERT, UPDATE_AFTER', 'u', 'DELETE', 'd' ] ); ``` #### Deletion-flag pattern Some consumers expect a boolean delete flag instead of an operation code. Map the row kinds to `true` and `false` in a column you name `deleted`: ```sql SELECT * FROM TO_CHANGELOG( input => TABLE my_aggregation, op => DESCRIPTOR(deleted), op_mapping => MAP[ 'INSERT, UPDATE_AFTER', 'false', 'DELETE', 'true' ] ); ``` #### Full versus partial deletes `produces_full_deletes` controls what a delete row carries. The *upsert key* is the column or columns that uniquely identify a row, taken from a declared primary key, a `GROUP BY` key, or the `PARTITION BY` key. With `produces_full_deletes => TRUE` (the default), each delete carries the full row image. For an upsert source that emits key-only deletes, Flink inserts a `ChangelogNormalize` step to rebuild the full image from state: ```sql -- Upsert source delete: -D[id:5, name:null] (key only) -- Output: +I[op:'DELETE', id:5, name:'Alice'] SELECT * FROM TO_CHANGELOG(input => TABLE upsert_source); ``` With `produces_full_deletes => FALSE`, Flink skips that step and emits deletes that carry only the key columns, with the rest set to `null`. This matches what an upsert sink such as a compacted topic expects, and it requires an upsert key or `PARTITION BY`: ```sql -- Output: +I[id:5, op:'DELETE', name:null] SELECT * FROM TO_CHANGELOG( input => TABLE upsert_source PARTITION BY id, produces_full_deletes => FALSE ); ``` #### NOTE With `produces_full_deletes => FALSE`, non-key columns are nulled on delete rows, so the output widens those columns to nullable even when they are `NOT NULL` on the input. Use the default if the output must keep the input’s `NOT NULL` types. This isn’t an Apache Kafka® tombstone. A delete row is still a real, fully serialized record with an explicit operation code (for example `+I[id:5, op:'DELETE', name:null]`), not a message with a `null` value. Kafka log compaction only treats a `null` message *value* as a tombstone, so writing this output to a compacted topic doesn’t by itself produce one. This shape is for a downstream consumer that reads the operation code and acts on it directly, which also makes it usable by a consumer that doesn’t understand Kafka’s tombstone convention at all. If you need an actual tombstone on delete, write an updating table’s `DELETE` row kind (for example, `FROM_CHANGELOG`’s output) directly to a sink with a `PRIMARY KEY`, instead of routing it through `TO_CHANGELOG` first. #### Partition by a key Use `PARTITION BY` when a downstream operator is keyed on that column. It moves the key columns to the front of the output: ```sql -- Input columns (name, id, cnt) become output columns (id, op, name, cnt) SELECT * FROM TO_CHANGELOG(input => TABLE my_aggregation PARTITION BY id); ``` ## Limitations The changelog functions map each input record to exactly one row kind. One input row can’t become two output rows, so a single message can’t be split into a full `UPDATE_BEFORE`/`UPDATE_AFTER` pair. A record that also happens to carry other fields, such as a before-image column you don’t reference, isn’t a problem by itself. It’s the 1:1 mapping that’s fixed, not the input’s shape. If your source sends both images in one message (for example, a single `MODIFY` event with both an old and new image), map it to `UPDATE_AFTER` alone using the new image and treat the result as an upsert stream instead of a retract stream. #### IMPORTANT `FROM_CHANGELOG` is an advanced feature. When you use it, you tell Flink that your stream is a valid changelog, and Flink doesn’t validate that claim. An incorrect changelog can produce silently wrong results downstream and, in the worst case, leave a statement in an unrecoverable state. When the data handed to `FROM_CHANGELOG` isn’t the shape the function expects, it becomes hard to diagnose what actually went wrong, so Confluent is only able to provide limited support for statements that contain `FROM_CHANGELOG`. ## Related content - [Process Table Functions](../../concepts/process-table-functions.md#flink-ptfs) - [Create a Process Table Function](../../how-to-guides/create-ptf.md#flink-ptfs-quickstart) - [Read and write custom changelog formats](../../how-to-guides/read-write-custom-changelog.md#flink-read-write-changelog) - [Changelog formats](../serialization.md#flink-sql-serialization-changelog-formats) #### 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).