Which Row Wins in Your Target

The question

Your CDC events arrive twice, or out of order. What does your warehouse or lakehouse do with them, and what do you have to set? CDC delivery is at-least-once, and end-to-end exactly-once across independent systems is not achievable in general. What you can reach is exactly-once processing (see effectively-once): at-least-once transport into an idempotent sink keyed on the primary key and ordered by the source log position (LSN, SCN or binlog coordinate), with deletes kept as delete markers.

Every target below has a way to say “this column decides which version wins”. Several default to something else: arrival order, commit order or a timestamp. The job is to name three things for your target: the key, the ordering column (the log position) and the delete marker.

Prerequisites: Event envelope, Materialization 101 and Making sinks bulletproof, which explains why a timestamp, a Kafka offset or ts_ms is the wrong version and how to compare positions. This page does not repeat that argument; it maps it onto six products.

Everything here was checked by reading each vendor’s documentation on 2026-10-09 (links at the end). Nothing was run against a live product. Where a behaviour is not stated in the pages read, the text says “check your version” instead of guessing.

The six targets at a glance

The pattern is the same everywhere: land the source position as a column, make it the ordering input, and keep the delete as a row that remembers its position. The rest of this page is the per-target detail.

A worked replay

One key, three source changes: position 100 sets name = 'a', position 200 sets name = 'b', position 300 deletes the key. Three things go wrong in practice: (1) the 100-batch is replayed after 200 was applied; (2) 200 arrives before 100; (3) 300 was applied, then 200 is replayed.

Replaying the same batch twice in order is harmless everywhere, because every row in it has the same position the second time. The danger is a replay that puts older content after newer content.

Why the vendors’ own examples are not the whole answer

Several vendor pages show a timestamp as the ordering input: BigQuery’s single-section example is a processing timestamp, Hudi’s SQL guide suggests a field “that denotes the time of the event”, Delta’s change-data example orders by a time column and the Debezium Server page lists __source_ts_ns for its Iceberg sink. Databricks’ multi-column example uses a timestamp and an ID to break ties. These examples are fine as illustrations of the setting, but a timestamp can tie for one key and can be skewed between hosts, so do not use one as the ordering key when the source gives you a log position. The log position is the safer key: it says which change committed later, and a replay carries the position it had the first time. Why, and how to compare positions: see Making sinks bulletproof. If your source has no total order per key (some non-relational sources), say so in the design and pick the best monotonic field it does give you.

Target by target

The snippets assume a staging table with id, op, a numeric source_lsn and an ordinal source_ordinal that separates changes sharing a position. All are marked untested.

BigQuery CDC ingestion

Default. You stream rows through the Storage Write API default stream with _CHANGE_TYPE set to UPSERT or DELETE. With identical primary keys the record most recently ingested by BigQuery wins. The default stream is documented as at-least-once, so duplicates are expected.

Failure with arrival order. A replayed old UPSERT is ingested later than the new one and overwrites it.

Safe configuration. Declare the primary key, then always set _CHANGE_SEQUENCE_NUMBER. It is a STRING of up to four hexadecimal sections separated by /, up to 16 hex characters each, compared section by section as unsigned numbers; the larger wins, and equal values fall back to ingestion time. Mixing rows with and without it gives unpredictable ordering. Put the log position first and the ordinal second, for example <lsn hex>/<ordinal hex>. A Postgres LSN printed as 16/B374D848 is already in that shape; that it compares in LSN order is my reading of the two pages, so test it.

Deletes. DELETE removes the row; BigQuery keeps delete operations for a two-day window so that out-of-order ones apply. A delete counts as applied only once upsert_stream_apply_watermark passes it. The page does not say what happens to an older UPSERT that arrives after that window: check. If you need the delete to be permanent, send an UPSERT with a deleted flag and the delete’s sequence number instead (a design choice, not a vendor feature). Tables with active CDC do not allow DML UPDATE, DELETE or MERGE, and key uniqueness is not enforced.

-- Untested: not run against a live BigQuery project.
CREATE TABLE analytics.customers (
  id INT64 PRIMARY KEY NOT ENFORCED,
  email STRING,
  deleted BOOL
)
CLUSTER BY id
OPTIONS (max_staleness = INTERVAL 10 MINUTE);

# Untested. Build the pseudocolumn values for each Storage Write API row.
def change_sequence_number(source_lsn: int, ordinal: int) -> str:
    # Two hex sections: the log position first, then the ordinal.
    return f"{source_lsn:X}/{ordinal:X}"

row = {
    "id": 7,
    "email": "a@example.com",
    "_CHANGE_TYPE": "UPSERT",  # "DELETE" for op = 'd'
    "_CHANGE_SEQUENCE_NUMBER": change_sequence_number(200, 0),
}

Databricks AUTO CDC

Default. There is none: KEYS and SEQUENCE BY are required, and the pipeline orders events by the sequencing column. The documentation asks for a column that is a monotonically increasing representation of the correct event order, with one distinct update per key at each value, and does not support NULL sequencing values. Its own sample drops an update at sequence 5 that arrives after one at 6.

Failure with a timestamp. Two changes to a key with the same timestamp break the “one distinct update per key at each value” requirement; what happens then is not documented: check. A skewed timestamp orders changes wrongly without any error.

Safe configuration. Sequence by the position, with the ordinal as the second field of a STRUCT (ordering is by the first field, then the second).

Deletes. APPLY AS DELETE WHEN treats an event as a delete. To handle out-of-order data the deleted row is retained as a tombstone for two days by default, configurable with pipelines.cdc.tombstoneGCThresholdInSeconds; the documentation says to set it above the longest delay between event arrival and pipeline execution. The SQL reference words this for SCD type 2 targets, so confirm it for SCD type 1 on your runtime. A replay older than the tombstone retention is not covered.

-- Untested: not run in a Lakeflow pipeline.
CREATE OR REFRESH STREAMING TABLE customers_current;

CREATE FLOW customers_apply AS AUTO CDC INTO customers_current
FROM stream(cdc.customers_changes)
KEYS (id)
APPLY AS DELETE WHEN op = 'd'
SEQUENCE BY STRUCT(source_lsn, source_ordinal)
COLUMNS * EXCEPT (op, source_lsn, source_ordinal)
STORED AS SCD TYPE 1;

Apache Hudi

Default. The merge mode is inferred. With no ordering field it is COMMIT_TIME_ORDERING: the record from the latest write wins, and the documentation says this expects input in strict order, with a database change log ordered by LSN as its example. With one or more ordering fields it is EVENT_TIME_ORDERING: the larger value wins, regardless of commit time. The table-level setting is hoodie.table.ordering.fields (the older precombine names are deprecated on the pages read; Flink uses ordering.fields). Do not change the merge mode after the table exists.

Failure with arrival order. A late old change overwrites a newer one under COMMIT_TIME_ORDERING. With an event-time field, a field that is a timestamp has the problems above.

Safe configuration. Make the ordering field the log position. Hudi Streamer’s Debezium guide already does this: --source-ordering-field _event_lsn for PostgreSQL, and a field derived from the binlog file and position for MySQL. Multiple ordering fields compare first to second, so list the position first and the ordinal second. The Hudi source (BufferedRecordMergerFactory on the master branch, which may differ from your release) shows why: shouldKeepNewerRecord keeps the newer record when ordering values are equal (compareTo(...) >= 0). With the LSN alone as ordering field, two changes that share an LSN therefore resolve by arrival order. The guide describes the payload comparison as applying when deduplicating within a batch, and the payload classes it uses are marked deprecated on the merger page, so check your version.

Deletes. A delete operation, or _hoodie_is_deleted = true, removes the record, and the Debezium payload applies d events as deletes. In the same source, deltaMergeDeleteRecord discards a delete whose ordering value is lower than the stored row’s as obsolete. A removed key leaves no marker, so my reasoning from that code is that a late, lower-ordered upsert can insert it again; the documentation pages do not say, so check your version, or keep your own deleted flag and position in the row.

-- Untested: not run on Spark with Hudi.
CREATE TABLE IF NOT EXISTS customers_hudi (
  id BIGINT,
  email STRING,
  deleted BOOLEAN,
  source_lsn BIGINT,
  source_ordinal BIGINT
) USING hudi
TBLPROPERTIES (
  type = 'mor',
  primaryKey = 'id',
  orderingFields = 'source_lsn,source_ordinal',
  recordMergeMode = 'EVENT_TIME_ORDERING'
);

Apache Iceberg

Default. An Iceberg table has no ordering column. Flink’s upsert mode (v2 table format, a primary key, write.upsert.enabled or upsert-enabled) writes row-level deletes by equality fields plus the new row. Under the spec, an equality delete applies only to data files with a strictly lower data sequence number, and sequence numbers are assigned per commit. My reading: the later commit wins, whatever position the data carried. The upsert section documents no version column.

Failure. A replay that commits after the newer change replaces it. In Spark, MERGE INTO errors if more than one source row can update a target row, so deduplicate the source first.

Safe configuration. Order before the sink, or merge with a guard on the position column and the ordinal, keyed on the primary key (snippet below). That guarded MERGE is the safe configuration.

The Debezium Server Iceberg sink is not a substitute. Its debezium.sink.iceberg.upsert-dedup-column has no default: it is optional in the sink’s IcebergConfig.java, and the documentation table on debezium.io, which lists __source_ts_ns, disagrees with the code and with the project’s own docs. In IcebergTableOperator.addToTable, deduplication runs per batch only, when upsert is on and the table has identifier fields. With the column blank, the last event of the batch for a key wins. With it set, the value is compared as a Long (null sorts lowest), and equal values fall back to operation priority d over u over r over c. Nothing in that code compares an event with the stored row, so an older value in a later batch overwrites a newer one: it gives no cross-batch protection. To put the position in the record, use the add.fields option of the event-flattening transform (lsn becomes __lsn), and set delete.tombstone.handling.mode=rewrite: without it a delete event lacks the added fields, so it has no __lsn and loses to any update in the batch. This is my reading of the repository code at master (memiiso/debezium-server-iceberg); check your version.

Deletes. Equality deletes are a v2/v3 feature; the draft v4 spec, which is not yet formally adopted, prohibits writing new ones, so check your format version and engine. The Debezium sink keeps deletes as __deleted = true rows by default (upsert-keep-deletes), which is the marker pattern.

-- Untested: not run on Spark with Iceberg.
-- stg is already one row per id (latest by source_lsn, source_ordinal).
MERGE INTO prod.db.customers t
USING stg s
ON t.id = s.id
WHEN MATCHED AND (t.source_lsn < s.source_lsn
     OR (t.source_lsn = s.source_lsn AND t.source_ordinal < s.source_ordinal)) THEN
  UPDATE SET t.email = s.email, t.deleted = (s.op = 'd'),
             t.source_lsn = s.source_lsn, t.source_ordinal = s.source_ordinal
WHEN NOT MATCHED THEN
  INSERT (id, email, deleted, source_lsn, source_ordinal)
  VALUES (s.id, s.email, (s.op = 'd'), s.source_lsn, s.source_ordinal);

Delta Lake: MERGE and change data feed

Default. Delta has no ordering setting; MERGE applies the source rows you give it. A merge can fail when several source rows match one target row, and the documentation tells you to reduce the source to the latest change per key first. Its change-data example does that with a time column (“can be replaced by other ordering id”), then applies deletes with whenMatchedDelete and inserts only rows that are not deleted. Nothing compares against the stored row.

Failure. With that example, a replayed older batch overwrites the newer row, and an older update after a delete finds no match and is inserted: the row is back.

Safe configuration. Dedupe the source by position, add a guard on the stored position, and keep the delete as a marker. In foreachBatch the documentation says the merge must be idempotent, because a restart can apply the same batch again.

Change data feed. The feed records row-level changes of a Delta table, with _change_type, _commit_version and _commit_timestamp. It is off by default, only captures changes after you enable it, and its files follow table retention, so VACUUM removes them. Its version is the Delta table’s own commit version, not a source position: a source of changes for tables downstream of this one, not a substitute for the source log.

-- Untested: not run on Delta Lake. stg is one row per id.
MERGE INTO customers t
USING stg s
ON t.id = s.id
WHEN MATCHED AND (t.source_lsn < s.source_lsn
     OR (t.source_lsn = s.source_lsn AND t.source_ordinal < s.source_ordinal)) THEN
  UPDATE SET t.email = s.email, t.deleted = (s.op = 'd'),
             t.source_lsn = s.source_lsn, t.source_ordinal = s.source_ordinal
WHEN NOT MATCHED THEN
  INSERT (id, email, deleted, source_lsn, source_ordinal)
  VALUES (s.id, s.email, (s.op = 'd'), s.source_lsn, s.source_ordinal);

Snowflake: MERGE and streams

Default. MERGE has no ordering. If one target row matches several source rows for an UPDATE or DELETE, the result is nondeterministic: with ERROR_ON_NONDETERMINISTIC_MERGE at its default of TRUE the merge errors; with FALSE one row is chosen and which one is not defined. A stream is a bookmark (an offset) on a table’s transactional versions that returns the changes since then, with METADATA$ACTION, METADATA$ISUPDATE and METADATA$ROW_ID. Put one over the table your loader lands CDC rows in and it orders by the landing table’s versions, not by the source log. Querying a stream does not advance it; consuming it in a DML statement does.

Failure. A stream over the landing table faithfully hands you the replayed duplicates and the swapped pair. An unguarded MERGE from it resurrects stale rows exactly as in the table above. The transactional stream consumption makes the landing-to-target hop safe to retry; it does nothing for the hop before it.

Safe configuration. Land the position as a column, read an append-only stream or the landing table, reduce to one row per key with QUALIFY ROW_NUMBER() by position, and merge with the guard and marker. The Snowflake lab shows the loading side.

-- Untested: not run on Snowflake. cdc_stream is an append-only stream
-- on the landing table cdc_landing(id, email, op, source_lsn, source_ordinal).
MERGE INTO customers t
USING (
  SELECT id, email, op, source_lsn, source_ordinal
  FROM cdc_stream
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY id ORDER BY source_lsn DESC, source_ordinal DESC) = 1
) s
ON t.id = s.id
WHEN MATCHED AND (t.source_lsn < s.source_lsn
     OR (t.source_lsn = s.source_lsn AND t.source_ordinal < s.source_ordinal)) THEN
  UPDATE SET email = s.email, deleted = (s.op = 'd'),
             source_lsn = s.source_lsn, source_ordinal = s.source_ordinal
WHEN NOT MATCHED THEN
  INSERT (id, email, deleted, source_lsn, source_ordinal)
  VALUES (s.id, s.email, (s.op = 'd'), s.source_lsn, s.source_ordinal);

A NULL stored position makes the guard NULL and the row is never updated: backfill it, or compare COALESCE(t.source_lsn, -1). Positions are comparable only within one source log lineage (see the lineage note in Making sinks bulletproof).

Delete markers, not physical deletes or tombstones alone

A delete in the log is a normal change event (op = 'd'). The Kafka tombstone that may follow is a compaction hint for the topic; sinks should act on the event, not wait for the tombstone. Whatever the target, a physical delete takes the version with it, so a replayed older update finds nothing to lose to. Either keep a marker row (a deleted flag plus the delete’s position) or rely on a target feature that keeps the delete for a stated time, and know the time: two days in BigQuery and Databricks by default. When the marker may finally go is the subject of Deletes That Stay Deleted.

Where exactly-once stops

Say it per hop. Database to Kafka: Debezium provides at-least-once delivery and has no deduplication layer of its own. Its documentation (from 3.3) describes an opt-in exactly-once mode for source connectors on Kafka Connect, built on KIP-618: Kafka Connect 3.3.0 or later in distributed mode, exactly.once.source.support=enabled on the workers and exactly.once.support=required on the connector. It covers that one hop, and the same page says it is unclear whether the underlying transaction implementation is fully correct, citing open Kafka issues. Kafka to your target: a separate system with no shared transaction. BigQuery CDC ingestion needs the default stream, which Google documents as at-least-once. The correctness you can claim at the target is exactly-once processing: duplicates and reordering happen, and the position-ordered idempotent write makes them harmless. See Exactly-once semantics.

Try it

Open the Change Feed Playground and pick the Re-insert after Update scenario (its id is snapshot-replay). In its event list the ledger row LED-100 is updated to version v2, then written again with its old v1 values, then v3 follows. That re-insert is an ordinary later source write, not a redelivered old event, so every target applies it, and a position guard would too. Write down what each target above would hold after the v1 write, then work out what would change if the same v1 record arrived a second time after v3: that is the case your default and the position-based setting disagree on. The Re-insert scenario cannot produce it, but two labs can, for one log lane and one key: Replay against a guarded sink delivers old changes again after newer ones, and ts_ms against log position shows a timestamp guard keeping the wrong change. They open the playground, a separate tool that contacts a third-party database (see the privacy note). The playground compares capture methods and models a guard in general terms; it does not simulate these six targets, so mapping what you see onto each target is on paper.

What to do Monday

1. For each target you load, write down the key, the ordering input and the delete representation. If the ordering input is arrival, commit order or a timestamp, that is your finding.
2. Carry the source position (and an ordinal for changes that share one) into the landing data. Compare it numerically, never as text.
3. Set it as the ordering input: _CHANGE_SEQUENCE_NUMBER, SEQUENCE BY, the Hudi ordering field, or the MERGE guard.
4. Keep deletes as markers with a position. Find out how long your target remembers a delete and compare it with how far back you can replay.
5. Replay one old batch and one swapped pair in staging and check the final rows against the table above.
6. Mark every “check your version” item above as an owner and a test, not an assumption.

Further resources

On this site

• Materialization 101: the same rule in generic SQL.
• Merge cookbook: merge templates for more engines.
• Offsets & replays: how replays happen.
• Snowflake CDC lab.
• Glossary: LSN / SCN, tombstone, idempotent write.

Vendor documentation (read 2026-10-09)

• BigQuery: stream table updates with CDC and the Storage Write API overview.
• Databricks: the AUTO CDC APIs, AUTO CDC INTO, create_auto_cdc_flow, change data capture and snapshots.
• Hudi: record merger (1.2.1), SQL DDL, writing data, Hudi Streamer (Debezium).
• Iceberg: Flink writes, Spark writes, table spec.
• Delta: update and merge, change data feed.
• Snowflake: streams, MERGE.
• Debezium: exactly-once delivery, Debezium Server (Iceberg sink), event flattening, and the Iceberg sink project docs.
• PostgreSQL: pg_lsn type.