# 1 Motivation
The dynamic table is a Flink SQL's fundamental abstraction, backed by a continuously evolving **changelog stream** (a stream of **INSERT**, **UPDATE**, and **DELETE** operations). The DataStream API already exposes this stream/table duality through flexible methods like toChangelogStream() and fromChangelogStream().
This FLIP proposes to bring a similar functionality and flexibility of these methods adapted into Flink SQL by using built-in Process Table Functions (PTFs): TO_CHANGELOG and FROM_CHANGELOG. This allows SQL users to have explicit, fine-grained control over the interpretation and generation of changelog streams, supporting advanced use cases like:
1. ***FROM_CHANGELOG***: enables custom SQL connectors by handling custom Change Data Capture (CDC) formats.
2. ***TO_CHANGELOG***: enables to create a CDC append stream from a Flink Table. This is the first operator that makes it possible to change an retract/upsert stream back to append.
With the introduction of PTFs, we now can define built-in PTFs as common utilities to complement our SQL offering as we take advantage of that here.
# 2 Background
This section contains some base knowledge used in the FLIP.
## 2.1 Change Data Capture Basics
*This section partially copied from our [PTFs documentation](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/ptfs/#change-data-capture-basics)*.
Under the hood, tables in Flink's SQL engine are backed by changelogs. These changelogs encode CDC (Change Data Capture) information containing INSERT (+I), UPDATE_BEFORE (-U), UPDATE_AFTER (+U), or DELETE (-D) messages.
The existence of these flags in the changelog constitutes the **Changelog Mode** of a consumer or producer:
### 2.1.1 Append Mode {+I}
- All messages are insert-only.
- Every insertion message is an immutable fact.
- Messages can be distributed in an arbitrary fashion across partitions and processors because they are unrelated.
### 2.1.2 Upsert Mode {+I, +U, -D}
- Messages can contain updates leading to an updating table.
- Updates are related using a key (i.e. the upsert key).
- Every message is either an upsert or delete message for a result under the upsert key.
- Messages for the same upsert key should land at the same partition and processor.
- Deletions can contain only values for upsert key columns (i.e. partial deletes) or values for all columns (i.e. full deletes).
- The mode is also known as *partial image* in the literature because -U messages are missing.
### 2.1.3 Retract Mode {+I, -U, +U, -D}
- Messages can contain updates leading to an updating table.
- Every insertion or update event is a fact that can be "undone" (i.e. retracted).
- Updates are related by all columns. In simplified words: The entire row is kind of the key but duplicates are supported. For example: `+I['Bob', 42]` is related to `-D['Bob', 42]` and `+U['Alice', 13]` is related to `-U['Alice', 13]`.
- Thus, every message is either an insertion (+) or its retraction (-).
- The mode is known as *full image* in the literature.
### 2.1.4 Partial vs Full Deletes
Deletions in changelog streams can be either:
- **Partial deletes**: Contain only the key columns (upsert key). The downstream system must have stored the full row to process the deletion.
- **Full deletes**: Contain all columns of the row being deleted. The downstream system can process the deletion without prior state.
Partial deletes are common in upsert mode where only the key is needed to identify the row to delete. Full deletes are typical in retract mode where the complete row image is retracted.
# 3 Public Interfaces
We'll introduce FROM_CHANGELOG and TO_CHANGELOG.
```
<FROM_CHANGELOG>
SELECT * FROM FROM_CHANGELOG(input => TABLE cdc_stream)
Complete signature:
SELECT * FROM FROM_CHANGELOG(
input => TABLE cdc_stream PARTITION BY partition_key ORDER BY timestamp,
uid => 'my-cdc-ptf',
before => DESCRIPTOR(before),
after => DESCRIPTOR(after),
op => DESCRIPTOR(op),
state_ttl => INTERVAL '10' MINUTE,
consume_full_deletes => false,
op_mapping => MAP[
'c', 'INSERT',
'd', 'DELETE',
'u', 'UPDATE_BEFORE, UPDATE_AFTER'
],
invalid_op_handling => 'FAIL'
);
<TO_CHANGELOG>
SELECT * FROM TO_CHANGELOG(input => TABLE t)
Complete signature:
SELECT * FROM TO_CHANGELOG(
input => TABLE t PARTITION BY k,
uid => 'my-cdc-ptf',
before => DESCRIPTOR(before),
after => DESCRIPTOR(after),
op => DESCRIPTOR(op),
state_ttl => INTERVAL '10' MINUTE,
produces_full_deletes => false,
op_mapping => MAP[
'INSERT', 'c',
'DELETE', 'd',
'UPDATE_BEFORE, UPDATE_AFTER', 'u'
]
)
-- Alternative for op_mapping param
op_mapping => MAP[
ARRAY['INSERT'], ARRAY['c'],
ARRAY['DELETE'], ARRAY['d'],
ARRAY['UPDATE_BEFORE', 'UPDATE_AFTER'], ARRAY['u']
]
```
# 4 Proposed Changes
We will introduce TO_CHANGELOG and FROM_CHANGELOG as PTFs, supporting the following signatures and semantics.
|