A Flink 2.3 INSERT INTO fails at planning because the query's upsert key differs from the sink's primary key - what is the risk, and how do you resolve it?

## Upsert key versus primary key Flink's planner derives an **upsert key** for every updating result: a set of columns that identifies each result row, so that every change targets exactly one row. For a `GROUP BY`, the upsert key is the grouping columns. A sink declares its own identity with `PRIMARY KEY (...) NOT ENFORCED`. When the sink key **contains** the query's upsert key, each result row maps to one stored row and nothing extra is needed. When it does not, several result rows can map to the same stored row - a **conflict** at the sink. ## How the mismatch happens without a join ```sql CREATE TABLE user_views ( user_id STRING, country STRING, views BIGINT, PRIMARY KEY (user_id) NOT ENFORCED ) WITH ('connector' = 'upsert-kafka', ...); INSERT INTO user_views SELECT user_id, country, COUNT(*) FROM page_views GROUP BY user_id, country; ``` The query's upsert key is `(user_id, country)`; the sink's key is `user_id`. When alice views pages from two countries, the result has two live rows, `alice/DE` and `alice/FR`, and both write key `alice`. The stored value is whichever change arrives last, and changes from different upstream keys can reach the sink through different parallel paths, so that order is not deterministic. With an updating input it gets worse: retracting `alice/DE` can delete `alice` in the sink although `alice/FR` still exists. ## What Flink did before 2.3 Under `table.exec.sink.upsert-materialize` (values `NONE`, `AUTO`, `FORCE`; default `AUTO`) the planner inserted a **`SinkUpsertMaterializer`** in front of an upsert sink whenever the key mismatch could reorder changes. It kept the history of rows per primary key in state so that it could restore the right surviving row on retraction. Planning succeeded, the plan showed `upsertMaterialize=[true]`, and the state could grow without a visible warning - a classic cause of a sink operator whose state and checkpoint size kept climbing. ## What Flink 2.3 requires `table.exec.sink.require-on-conflict` defaults to `true`: if the upsert key differs from the sink's primary key and the statement has no `ON CONFLICT` clause, planning fails with a message that names both keys. The clause goes at the end of the `INSERT`: | Strategy | On a real conflict | State | Needs source watermarks | |---|---|---|---| | `ON CONFLICT DO ERROR` | fails the job at runtime | buffered changes, compacted on watermark | yes | | `ON CONFLICT DO NOTHING` | keeps the first record, discards later ones | buffered changes, compacted on watermark | yes | | `ON CONFLICT DO DEDUPLICATE` | keeps the latest, rolls back correctly on retraction | full history per key | no | `DO ERROR` and `DO NOTHING` buffer changes by primary key and upsert key and compact them when the watermark advances, so that a `-U` and a `+I` arriving out of order do not look like a conflict. Planning rejects them if any source table lacks a `WATERMARK` declaration. ## Deciding what to do 1. **Look for a logic error first.** The check exists to surface it. Here either the query should `GROUP BY user_id` only, or the sink should declare `PRIMARY KEY (user_id, country)`. 2. **If the keys are logically equal but the planner cannot prove it**, use `DO ERROR`: it costs little and fails loudly if you were wrong. 3. **If dropping later rows is acceptable**, use `DO NOTHING`. 4. **If several upstream rows genuinely update one key and correctness matters**, use `DO DEDUPLICATE`, and budget and monitor its state. ## Symptoms of the mismatch in production Before the 2.3 check, and still when the check is switched off, the mismatch shows up indirectly: - a sink value that flips between two versions for the same key from one run to the next, because the last writer depends on scheduling; - a key that disappears from the sink although the source still has live rows for it, because a retraction of one upstream row deleted the shared key; - a `SinkUpsertMaterializer` operator whose state and checkpoint size grow steadily with traffic rather than with the number of keys. Each of these points back to the same question: which columns identify a row of the result, and does the sink agree? ## Escape hatches and their cost - `table.exec.sink.require-on-conflict` = `false` restores the pre-2.3 behaviour: no clause required, and results may be non-deterministic. - `table.exec.sink.upsert-materialize` = `NONE` removes the materializer: no buffering, no compaction, no conflict resolution; records go straight to the sink in arrival order. - Neither setting fixes the underlying mismatch; both only decide who notices it and when.