Flink SQL Evolution: Handling Custom CDC with Process Table Functions FLIP-564: FROM_CHANGELOG and TO_CHANGELOG built-in PTFs Gustavo de Morais OCTOBER 11-14, 2026 GLASGOW, SCOTLAND

Introduction • Gustavo de Morais • Apache Flink committer • Staff Software Engineer at Confluent, an IBM company • MultiJoin (FLIP-516), FROM/TO_CHANGELOG (FLIP-564), planner and engine optimizations and fixes including (FLIP-568) October 11-14, 2026 Glasgow, Scotland

Who is Confluent, an IBM company? • Enterprise Kafka, Flink and Iceberg. Built on top of open source • One of the main contributors to Apache Flink • 105 of 372 accepted FLIPs (30%) October 11-14, 2026 Glasgow, Scotland

What is Apache Flink? • The de facto standard for stream processing • True streaming architecture: 1ms to 1s • Three main APIs: Flink SQL, Table API, DataStream October 11-14, 2026 Glasgow, Scotland

Since Flink 2.0, contributors have been building in Table and SQL Table & SQL flink-table DataStream API flink-streaming-java DataStream API V2 flink-datastream(-api) 106 contributors · 774 commits 32 contributors · 58 commits 3 contributors · 3 commits Bar length: contributors. git log —no-merges, apache/flink, 1 Mar 2025 – 7 Oct 2026. flink-streaming-java also contains runtime code used by SQL jobs. October 11-14, 2026 Glasgow, Scotland

Table, CDC and Changelogs From tables to changes OCTOBER 11-14, 2026 GLASGOW, SCOTLAND

Table, CDC and Changelogs 01 02 03 Table CDC Changelog the current state. captures every change to that state. the stream of changes. October 11-14, 2026 Glasgow, Scotland

Bank statement analogy The balance is the table. The statement is the changelog. Account Account Balance Checking 100 Statement Description Amount Deposit +130 Deposit +20 Withdrawal -50 October 11-14, 2026 Glasgow, Scotland

What are ProcessTableFunctions (PTFs)? OCTOBER 11-14, 2026 GLASGOW, SCOTLAND

Flink SQL supports your own user-defined operator … SELECT * FROM MyCustomLogic( TABLE Events PARTITION BY name ) … October 11-14, 2026 Glasgow, Scotland

What is a ProcessTableFunction in Flink SQL? A user-defined function with access to (almost) everything a built-in operator has Timers State remember earlier rows October 11-14, 2026 continue at a later point in time Watermarks work in event time Changelog consume and produce updates Multiple inputs Ordering Partitioning several tables in one function rows in order within a partition one instance of your logic per key Glasgow, Scotland

@DataTypeHint(“ROW<message STRING>”) public class GreetingWithMemory extends ProcessTableFunction<Row> implements ChangelogFunction { public static class CountState { public long counter = 0L; } public ChangelogMode getChangelogMode(ChangelogContext ctx) { return ChangelogMode.upsert(); } public void eval(@StateHint CountState state, @ArgumentHint(SET_SEMANTIC_TABLE) Row input) { state.counter++; RowKind kind = state.counter > 1 ? UPDATE_AFTER : INSERT; collect(Row.ofKind(kind, “Hello ” + input.getFieldAs(“name”) + “, your ” + state.counter + ” time?”)); } } October 11-14, 2026 Glasgow, Scotland

Use it in SQL and Table API SQL Table API SELECT * FROM GreetingWithMemory( TABLE Customer PARTITION BY name ); env.from(“Customer”) .partitionBy($(“name”)) .process(GreetingWithMemory.class); Changes emitted for the customers Bob, Alice, Bob op name message INSERT Bob Hello Bob, your 1 time? INSERT Alice Hello Alice, your 1 time? UPDATE_AFTER Bob Hello Bob, your 2 time? October 11-14, 2026 Glasgow, Scotland

Flink docs Everything about PTFs, with runnable examples, is on one page https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/ptfs/ #process-table-functions-ptfs October 11-14, 2026 Glasgow, Scotland

Suggestion October 11-14, 2026 Glasgow, Scotland

Handling CDC in Flink SQL with ProcessTableFunctions: TO and FROM_CHANGELOG OCTOBER 11-14, 2026 GLASGOW, SCOTLAND

1st Use case: CDC & Changelog in the real world with replication CDC stream DB 1 CDC stream Flink DB 2 source copy user age user age ana 31 ana 31 October 11-14, 2026 Glasgow, Scotland

CDC & Changelog in the real world DynamoDB Streams Debezium {“eventName”: “INSERT”, … “NewImage”: {“user”: {“S”:”ana”}, “age”: {“N”:”30”}}} {“op”: “c”, “before”: null, “after”: {“user”: “ana”, “age”: 30}} {“eventName”: “MODIFY”, … “OldImage”: {“user”: {“S”:”ana”}, “age”: {“N”:”30”}}, “NewImage”: {“user”: {“S”:”ana”}, “age”: {“N”:”31”}}} {“op”: “u”, “before”: {“user”: “ana”, “age”: 30}, “after”: {“user”: “ana”, “age”: 31}} {“eventName”: “REMOVE”, … “OldImage”: {“user”: {“S”:”ana”}, “age”: {“N”:”31”}}} October 11-14, 2026 {“op”: “d”, “before”: {“user”: “ana”, “age”: 31}, “after”: null} Glasgow, Scotland

One change, three dialects DynamoDB Streams DB 1 Debezium DB 2 Flink DynamoDB MySQL DynamoDB Streams Flink changelog Debezium INSERT INSERT (+I) op: c MODIFY, old image UPDATE BEFORE (-U) op: u, before MODIFY, new image UPDATE AFTER (+U) op: u, after REMOVE DELETE (-D) op: d October 11-14, 2026 Glasgow, Scotland

Until now, only possible with 01 02 03 Native connectors Flink CDC DataStream API the Debezium, Canal, Maxwell and Ogg formats, upsert-kafka connectors for popular databases your own deserializer, in Java October 11-14, 2026 Glasgow, Scotland

Flink SQL Native Solution Changelog any format FROM_CHANGELOG Flink table any SQL TO_CHANGELOG Changelog any format Any changelog in, any SQL in the middle, any changelog out October 11-14, 2026 Glasgow, Scotland

Flink SQL Native Solution: the stream as a table dynamodb_stream — the real records {“eventName”: “INSERT”, … “NewImage”: {“user”:{“S”:”ana”},”age”:{“N”:”30”}}} {“eventName”: “MODIFY”, … “OldImage”: {“user”:{“S”:”ana”},”age”:{“N”:”30”}}, “NewImage”: {“user”:{“S”:”ana”},”age”:{“N”:”31”}}} {“eventName”: “REMOVE”, … “OldImage”: {“user”:{“S”:”ana”},”age”:{“N”:”31”}}} CREATE VIEW dynamodb_cdc AS SELECT eventName, COALESCE(dynamodb.NewImage.user.S, dynamodb.OldImage.user.S) AS user, CAST(COALESCE(dynamodb.NewImage.age.N, dynamodb.OldImage.age.N) AS INT) AS age FROM dynamodb_stream; October 11-14, 2026 Glasgow, Scotland

In pure SQL, step 1: FROM_CHANGELOG dynamodb_cdc — in +I[INSERT, ana, 30] +I[MODIFY, ana, 31] +I[REMOVE, ana, 31] users — the Flink table +I[ana, 30] +U[ana, 31] -D[ana, 31] CREATE VIEW users AS SELECT * FROM FROM_CHANGELOG( input => TABLE dynamodb_cdc PARTITION BY user, op => DESCRIPTOR(eventName), op_mapping => MAP[‘INSERT’, ‘INSERT’, ‘MODIFY’, ‘UPDATE_AFTER’, ‘REMOVE’, ‘DELETE’]); October 11-14, 2026 Glasgow, Scotland

In pure SQL, step 2: TO_CHANGELOG users — the Flink table +I[ana, 30] +U[ana, 31] -D[ana, 31] debezium_cdc — out +I[c, ana, 30] +I[u, ana, 31] +I[d, ana, 31] INSERT INTO debezium_cdc SELECT * FROM TO_CHANGELOG( input => TABLE users, op => DESCRIPTOR(op), op_mapping => MAP[‘INSERT’, ‘c’, ‘UPDATE_AFTER’, ‘u’, ‘DELETE’, ‘d’]); October 11-14, 2026 Glasgow, Scotland

Flink SQL Native Solution: the same, in the Table API Same input and output as the SQL version Table users = tEnv.from(“dynamodb_cdc”) .partitionBy($(“user”)) .fromChangelog( descriptor(“eventName”).asArgument(“op”), map(“INSERT”, “INSERT”, “MODIFY”, “UPDATE_AFTER”, “REMOVE”, “DELETE”).asArgument(“op_mapping”)); users.toChangelog( descriptor(“op”).asArgument(“op”), map(“INSERT”, “c”, “UPDATE_AFTER”, “u”, “DELETE”, “d”).asArgument(“op_mapping”)) .insertInto(“debezium_cdc”).execute(); October 11-14, 2026 Glasgow, Scotland

Under the hood Using the functions sounds simple, right? Making this reliable, scalable, and efficient across billions of records between two systems isn’t. That’s what Flink takes care of under the hood. October 11-14, 2026 Glasgow, Scotland

FROM_CHANGELOG Append-only stream with an operation column becomes a table SELECT * FROM FROM_CHANGELOG( input => TABLE source_table PARTITION BY key_col [ORDER BY time_col], op => DESCRIPTOR(op_column_name), op_mapping => MAP[‘c, r’, ‘INSERT’, ‘u’, ‘UPDATE_AFTER’, ‘d’, ‘DELETE’], error_handling => ‘FAIL’ | ‘SKIP’ ); October 11-14, 2026 Glasgow, Scotland

TO_CHANGELOG Table becomes an append-only changelog stream SELECT * FROM TO_CHANGELOG( input => TABLE source_table [PARTITION BY key_col], op => DESCRIPTOR(op_column_name), op_mapping => MAP[‘INSERT, UPDATE_AFTER’, ‘u’, ‘DELETE’, ‘d’], produces_full_deletes => TRUE | FALSE ); October 11-14, 2026 Glasgow, Scotland

2nd Use cases: getting around the Flink engine! EXAMPLES! October 11-14, 2026 Glasgow, Scotland

Auditing your database CDC stream of a table as an append-only audit log INSERT INTO orders_audit_log SELECT * FROM TO_CHANGELOG( input => TABLE orders ); With an upsert source there is no UPDATE_BEFORE row. October 11-14, 2026 Input - orders (current state) id status 1 SHIPPED Output - orders_audit_log op id status INSERT 1 NEW UPDATE_BEFORE 1 NEW UPDATE_AFTER 1 SHIPPED INSERT 2 NEW DELETE 2 NEW Glasgow, Scotland

Most traded asset The most traded asset has the most changes CREATE VIEW asset_changes AS SELECT ticker, COUNT(*) AS changes FROM TO_CHANGELOG(input => TABLE assets) WHERE op <> ‘UPDATE_BEFORE’ GROUP BY ticker; SELECT * FROM asset_changes ORDER BY changes DESC LIMIT 1; AAPL has 2 changes, TSLA has 1. The most traded asset is AAPL. October 11-14, 2026 Input - assets (one row per asset) ticker price AAPL 191 TSLA 250 Output - changes per asset op ticker price INSERT AAPL 190 UPDATE_BEFORE AAPL 190 UPDATE_AFTER AAPL 191 INSERT TSLA 250 Glasgow, Scotland

Soft-delete flag No CDC tool at all: a deleted column becomes a real DELETE CREATE VIEW customers AS SELECT * FROM FROM_CHANGELOG( input => TABLE customers_raw PARTITION BY id, op => DESCRIPTOR(deleted), op_mapping => MAP[‘false’, ‘UPDATE_AFTER’, ‘true’, ‘DELETE’] ); Downstream queries see a real delete. Bob is gone, not flagged. October 11-14, 2026 Input - customers_raw id name deleted 1 alice false 2 bob false 1 alice smith false 2 bob true Output - customers id name 1 alice smith Glasgow, Scotland

Beloved error When an operator or sink cannot consume the changes it is given org.apache.flink.table.api.TableException: Can’t generate a valid execution plan for the given query. There is a changelog mismatch between two operators. One produces an upsert changelog (UPDATE_AFTER without UPDATE_BEFORE). The other requires a retract changelog (UPDATE_BEFORE and UPDATE_AFTER), for example a sink without a primary key. … The conflict is at: Sink(expectedChangelogMode=[I,UB,UA,D]) +- ProcessTableFunction(changelogMode=[I,UA,D]) October 11-14, 2026 Glasgow, Scotland

LAG on an updating table LAG only accepts append-only input, so turn the updates into inserts first INSERT INTO sink SELECT order_id, op, status, LAG(status) OVER (PARTITION BY order_id ORDER BY ts) AS prev_status FROM TO_CHANGELOG(input => TABLE orders); sink — the previous status on every change order_id op status prev_status 1 INSERT NEW NULL 1 UPDATE_AFTER SHIPPED NEW 1 UPDATE_AFTER DELIVERED SHIPPED October 11-14, 2026 Glasgow, Scotland

Many more • Operators that need append-only input, e.g. LAG over OVER • TO_CHANGELOG turns the updates into inserts so they work • Append-only sinks • Deduplicating records without watermarks • Kafka tombstones as a side pipeline • Found another use? I would love to hear about it October 11-14, 2026 Glasgow, Scotland

What can go wrong? • Order matters: events must arrive in order per key • Unknown operation codes fail the job, or vanish with SKIP • A wrong op_mapping silently gives a wrong table • Partial deletes keep only the key columns • Upsert output needs a sink that accepts upserts October 11-14, 2026 Glasgow, Scotland

What can go wrong? • Order matters: events must arrive in order per key • Unknown operation codes fail the job, or vanish with SKIP ! g n i h Not orks • A wrong op_mapping silently gives a wrong table • Partial deletes keep only a w ayscolumns lwkey ink the Fl • Upsert output needs a sink that accepts upserts October 11-14, 2026 Glasgow, Scotland

What can go wrong? op_mapping => MAP[ ‘INSERT’, ‘DELETE’, ‘DELETE’, ‘INSERT’ ] October 11-14, 2026 Glasgow, Scotland

Sum up 01 02 03 Process table functions Bring your own CDC More than replication FROM_CHANGELOG in, TO_CHANGELOG out. audit logs, soft deletes, append-only input for operators and sinks. your own operator in Flink SQL. State, timers and changelogs included. October 11-14, 2026 Built in, from SQL or the Table API. Glasgow, Scotland

What is next? • Flink 2.3: very limited support • Flink 2.4: everything in this talk • Not implemented: splitting and merging update events (previous and new image) October 11-14, 2026 Glasgow, Scotland

Adoption so far Jul 13 Jul 20 Jul 27 Aug 03 Aug 10 Customers active that week October 11-14, 2026 Aug 17 Aug 24 Aug 31 Running total of customers Sep 07 Sep 14 Sep 21 Sep 28 Running in production CUSTOMERS TO vs FROM +24% 94% / 6% growth week over week of statements Glasgow, Scotland

Resources Blog post, FLIP-564 and the Flink docs, all in one place https://gustavodemorais.github.io/glasgow Scan to open October 11-14, 2026 Glasgow, Scotland

THANK YOU - QUESTIONS? github.com/gustavodemorais linkedin.com/in/gustavo-demorais