Definition
- Append-only – Contains only insert rows, never updates or deletes, fitting insert-only sources like a raw Kafka event feed.
- Retract – Encodes updates as pairs: a minus row retracting the old value and a plus row adding the new value.
- Upsert – Keyed stream where an upsert replaces any prior row with the same key, more compact than retract for keyed updates.
- Changelog mode matching – The planner derives the query's changelog mode and rejects plans where the sink cannot consume it.
- CDC fit – Change-data-capture feeds are naturally upsert streams keyed by the database primary key.
Think of how the National Bank of India communicates changes. An append-only stream is like a transaction journal where you only ever add new lines and never erase. A retract stream is like telling the clerk "cancel the old total of one thousand, and record a new total of one thousand two hundred fifty." An upsert stream is like handing the clerk a card stamped with an account number and saying "file this as the current balance, replacing whatever card was there before."
Understanding the Concepts
What is an append-only changelog stream?
An append-only stream contains a single row kind: insert. It arises from insert-only sources and from queries that never change previously emitted rows, like a filter, a projection, or a windowed aggregation that emits a final result once per window. Append-only is the simplest and cheapest mode because downstream systems never have to revise anything.
-- Append-only source: raw Kafka transaction feed
CREATE TABLE raw_transactions (
txn_id STRING,
account_id STRING,
amount DECIMAL(15, 2),
txn_time TIMESTAMP(3),
WATERMARK FOR txn_time AS txn_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-raw-transactions',
'format' = 'json'
);
-- Append-only query: simple filter (no updates)
SELECT * FROM raw_transactions WHERE amount > 1000000;
-- Every row is final. Any sink can consume this.
What is a retract changelog stream?
A retract stream encodes changes as two row kinds: a plus row that adds a value and a minus row that retracts a previously added value. When a non-windowed aggregation changes a result, Flink emits a minus row for the old aggregate and a plus row for the new one.
-- Running count per account produces a retract stream
SELECT account_id, COUNT(*) AS txn_count
FROM raw_transactions
GROUP BY account_id;
-- As transactions arrive:
-- txn-001 (ACCT-100) → +I (ACCT-100, 1) ← insert
-- txn-002 (ACCT-100) → -U (ACCT-100, 1) ← retract old
-- +U (ACCT-100, 2) ← insert new
-- txn-003 (ACCT-200) → +I (ACCT-200, 1) ← insert
-- txn-004 (ACCT-100) → -U (ACCT-100, 2) ← retract old
-- +U (ACCT-100, 3) ← insert new
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class RetractStreamDemo {
public static void main(String[] args) {
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
// This aggregation produces retract messages
// The sink must handle minus rows
tEnv.executeSql(
"SELECT account_id, SUM(amount) AS total " +
"FROM raw_transactions GROUP BY account_id"
);
}
}
What is an upsert changelog stream?
An upsert stream is keyed. It carries upsert rows (insert-or-update for a key) and delete rows. Because the key identifies which row to replace, an update is a single message, more compact than retract's two messages.
-- Upsert sink: account totals keyed by account_id
CREATE TABLE account_totals (
account_id STRING,
total_amount DECIMAL(15, 2),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'nbi-account-totals',
'key.format' = 'json',
'value.format' = 'json'
);
-- This aggregation produces an upsert stream keyed by account_id
INSERT INTO account_totals
SELECT account_id, SUM(amount)
FROM raw_transactions
GROUP BY account_id;
-- Each update is ONE message that replaces the prior value:
-- ACCT-100: 50000 → single upsert message
-- ACCT-100: 80000 → single upsert message (replaces 50000)
When does each changelog mode apply?
| Changelog Mode | Row Kinds | When It Applies | Example Query |
|---|---|---|---|
| Append-only | INSERT only | Filters, projections, window aggregations | SELECT * WHERE amount > 1M |
| Retract | +I, -U, +U, -D | Non-windowed aggregations, non-keyed updates | SELECT acct, COUNT(*) GROUP BY acct |
| Upsert | +U, -D (keyed) | CDC feeds, keyed aggregations with primary key | INSERT INTO upsert_sink SELECT ... |
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class ChangelogModeComparison {
public static void main(String[] args) {
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
// Append-only: filter only
tEnv.executeSql("INSERT INTO alerts SELECT * FROM transactions WHERE amount > 1000000");
// Retract: aggregation without keyed sink
tEnv.executeSql("SELECT branch_id, AVG(amount) FROM transactions GROUP BY branch_id");
// Upsert: aggregation into keyed sink
tEnv.executeSql("INSERT INTO branch_totals SELECT branch_id, SUM(amount) FROM transactions GROUP BY branch_id");
}
}
Why must the changelog mode match the sink?
The planner derives the changelog mode from the query's semantics. An append-only query can go to any sink. But an updating query that produces retract or upsert messages cannot be written to a plain append-only sink. The planner will reject that plan.
-- This FAILS: updating aggregation → append-only sink
CREATE TABLE append_sink (
account_id STRING,
total DECIMAL(15, 2)
) WITH ('connector' = 'filesystem', 'path' = '/data/output', 'format' = 'csv');
INSERT INTO append_sink
SELECT account_id, SUM(amount) FROM raw_transactions GROUP BY account_id;
-- ERROR: the query produces retract messages but the sink only accepts appends
-- FIX: use an upsert-capable sink
CREATE TABLE upsert_sink (
account_id STRING,
total DECIMAL(15, 2),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH ('connector' = 'upsert-kafka', 'topic' = 'nbi-totals',
'key.format' = 'json', 'value.format' = 'json');
How does CDC fit into changelog modes?
CDC feeds are naturally upsert streams keyed by the database primary key. Each database change carries an operation type: insert, update, or delete, which maps directly to upsert changelog semantics.
-- CDC source: upsert stream from MySQL binlog
CREATE TABLE accounts_cdc (
account_id STRING,
balance DECIMAL(15, 2),
status STRING,
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'mysql.bank.internal',
'database-name' = 'nbi',
'table-name' = 'accounts'
);
-- Each change is an upsert keyed by account_id
-- INSERT → new account opened
-- UPDATE → balance changed
-- DELETE → account closed
What are common mistakes?
The most common mistake is pointing an updating query at an append-only sink. The second is not declaring a primary key on an upsert sink.
-- WRONG: no primary key on upsert sink
CREATE TABLE bad_sink (
account_id STRING,
total DECIMAL(15, 2)
) WITH ('connector' = 'upsert-kafka', 'topic' = 'nbi-totals',
'key.format' = 'json', 'value.format' = 'json');
-- ERROR: upsert-kafka requires a PRIMARY KEY declaration
-- CORRECT: declare primary key
CREATE TABLE good_sink (
account_id STRING,
total DECIMAL(15, 2),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH ('connector' = 'upsert-kafka', 'topic' = 'nbi-totals',
'key.format' = 'json', 'value.format' = 'json');
What should a senior engineer know?
- Upsert is more compact than retract for keyed updates because it sends one message per update instead of two.
- Converting a retract or upsert stream to append-only requires choosing a final value, only possible for bounded or windowed results.
- Window aggregations emit final results after the window closes, so they are append-only.
- Monitor the retract ratio in your pipeline: a high retract rate on a non-keyed query may signal you should restructure to use an upsert sink.
// Senior tip: use EXPLAIN to check changelog mode
tEnv.executeSql(
"EXPLAIN SELECT account_id, SUM(amount) FROM transactions GROUP BY account_id"
);
// Output shows: changelog mode = [I, -U, +U]
// This means retract mode: the sink must support it
Visual Overview
Three Changelog Modes Append-Only: Retract: Upsert: ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +I (A, 100) │ │ +I (A, 100) │ │ +U (A, 100) │ │ +I (B, 200) │ │ -U (A, 100) │ │ +U (A, 150) │ │ +I (A, 50) │ │ +U (A, 150) │ │ +U (B, 200) │ │ │ │ +I (B, 200) │ │ -D (A) │ └──────────────┘ └──────────────┘ └──────────────┘ Only inserts Pairs for updates Keyed, single msg Any sink works Retract-capable sink Upsert sink + PK
Query → Changelog Mode → Sink Matching ┌─────────────────┐ ┌──────────────┐ ┌──────────────────┐ │ Filter/Project │ ──▶ │ Append-only │ ──▶ │ Any sink │ └─────────────────┘ └──────────────┘ └──────────────────┘ ┌─────────────────┐ ┌──────────────┐ ┌──────────────────┐ │ GROUP BY (no key)│ ──▶ │ Retract │ ──▶ │ Retract-capable │ └─────────────────┘ └──────────────┘ └──────────────────┘ ┌─────────────────┐ ┌──────────────┐ ┌──────────────────┐ │ GROUP BY + PK │ ──▶ │ Upsert │ ──▶ │ Upsert sink + PK │ └─────────────────┘ └──────────────┘ └──────────────────┘
Key Takeaway
In Flink SQL every dynamic table is produced and consumed as a changelog stream, and there are three modes. Append-only contains only insert rows and never revises anything, fitting filters, projections, and window aggregations. Retract encodes updates as a minus row for the old value and a plus row for the new value, used by non-windowed aggregations. Upsert is keyed and sends one compact message per update, which is the natural shape for CDC feeds and aggregations written to a keyed sink.
The decisive practical point is that the planner derives the changelog mode from the query and the mode must match the sink. An append-only query goes to any sink, but an updating aggregation needs an upsert-capable or retract-capable sink. Pointing an updating query at an append-only sink is a compilation error, and forgetting the primary key on an upsert sink is the second most common mistake.
At the National Bank of India we hit all three. A filter for high-value alerts was append-only and we wrote it anywhere. A per-account running count was retract, so we used a retract-capable sink. Our CDC account-balance feed was a keyed upsert stream written through the upsert-kafka connector with a declared primary key. Understanding which mode a query produces and matching it to the sink is exactly what separates someone who has shipped Flink SQL from someone who has not.
Remember one sentence: append-only emits inserts only, retract sends cancel-and-replace pairs, upsert sends one keyed message per update, and the query's changelog mode must match the sink or the planner rejects the plan.
---