# Deduplication and merge Turning raw records, with their duplicates and change events, into one correct row per entity: choosing the winning record for each key and applying it to the clean table with MERGE. > [!NOTE] > > **Before reading** > > [Change data capture]() ยท [Delivery guarantees and idempotency]() ## Why it matters The raw layer holds every delivery, including repeats, and every change, including deletes. Reports need the current state: one row per order, with deleted orders gone. This step is where the platform's numbers become right or wrong, and it must give the same result however often it runs. ## How it works Three questions decide the logic. What identifies an entity? Usually the source's primary key, such as order\_id. For events, it is the event ID. If neither exists, a key has to be built from several columns, and that choice must be agreed with people who know the data. What makes one record newer than another? The best answer is a sequence from the source: a log position from CDC, a Kafka offset within one key, or a version number the source increments. Timestamps are a weaker choice. Two changes can share a millisecond, clocks differ between servers, and a retried event can carry an older time than a later one. When only timestamps exist, add a tie-breaker so the result is deterministic. What happens to deletes? With CDC, a delete is a change event like any other. If the latest event for a key is a delete, the key must be removed from, or marked deleted in, the clean table. The usual implementation has two steps. First, rank the records for each key and keep the latest one, with a window function. Second, merge that result into the clean table: update keys that exist, insert keys that do not, and delete keys whose latest event is a delete. ``` flowchart LR R[(raw_orders:
o-1001 created
o-1001 updated
o-1001 updated, resent
o-1002 created)] --> L[Latest per order_id
by source sequence] L --> M{MERGE on order_id} M -->|exists, not deleted| U[Update] M -->|new| I[Insert] M -->|latest is delete| D[Delete] U --> C[(clean_orders)] I --> C D --> C ``` The merge reads only the raw rows added since the last run, which keeps it cheap. It must still be safe to repeat: ranking first guarantees one source row per key, and comparing sequences in the merge condition prevents an older change from overwriting a newer one if the same rows are processed twice or out of order. Deduplication of exact repeats, the same event delivered twice, falls out of the same logic, because both copies have the same key and sequence and one is kept. Rebuilding from scratch is the other option: rank all raw rows and write the result as a new table. It is simpler and always correct, but it reads the whole raw history on every run. Many teams merge incrementally every day and rebuild completely when logic changes. ## Key terms - Business key The identifier of an entity in the source, such as order\_id. - Winning record The record chosen as current for a key. - Source sequence An ordering from the source, such as a log position or offset. - Tie-breaker An extra ordering column that makes the choice deterministic. - MERGE One statement that inserts, updates and deletes rows in a target based on a source. - Soft delete Marking a row as deleted instead of removing it; it keeps the stored sequence, so a replay cannot bring a deleted row back. ## Example raw\_orders is filled by CDC through the Iceberg sink, with the operation and Kafka offset in `_cdc` (the sink's DebeziumTransform maps the Debezium operations c, u and d to I, U and D). Building clean\_orders from the rows ingested since the last run: ``` MERGE INTO shop.clean_orders AS t USING ( SELECT * FROM ( SELECT r.*, row_number() OVER (PARTITION BY order_id ORDER BY r._cdc.offset DESC) AS rn FROM shop.raw_orders AS r WHERE r._ingested_at >= TIMESTAMP '2026-10-28 00:00:00 UTC' ) WHERE rn = 1 ) AS s ON t.order_id = s.order_id WHEN MATCHED AND s._cdc.op = 'D' AND s._cdc.offset > t.source_offset THEN DELETE -- the sink writes ops as I, U, D WHEN MATCHED AND s._cdc.offset > t.source_offset THEN UPDATE SET status = s.status, amount = s.amount, updated_at = s.updated_at, source_offset = s._cdc.offset WHEN NOT MATCHED AND s._cdc.op <> 'D' THEN INSERT (order_id, customer_id, status, amount, ordered_at, updated_at, source_offset) VALUES (s.order_id, s.customer_id, s.status, s.amount, s.ordered_at, s.updated_at, s._cdc.offset); ``` Order o-1001 has three raw rows, one of them a resend. Only the latest by offset reaches the merge. If the job runs twice, the second run changes nothing, because the stored source\_offset is not lower than the incoming one. This example removes deleted orders. Because the stored source\_offset goes with them, a later replay of older raw rows for that order would insert it again; a soft delete (an is\_deleted flag, keeping source\_offset) prevents that, and consumers filter on the flag. ## Pitfalls - Ordering by a timestamp with ties. The winning row changes between runs; add a tie-breaker. - Duplicate keys in the merge source. Rank first, so each key appears once. - Dropping deletes. Deleted orders then remain in revenue. - Merging without a sequence check. Reprocessing old rows overwrites newer data. - An incremental window that misses late rows. Choose the window by ingestion time, not event time, and overlap it with the previous run by at least the sink's commit interval; the sequence check makes the overlap harmless. ## Learn more - [Trino documentation, MERGE]() - [Apache Iceberg, Spark writes (MERGE INTO)]() ## Related [History and slowly changing dimensions]() [Backfill and replay]() [Writing to Iceberg from Kafka]() [dbt]()