Delta MERGE fails with 'multiple source rows matched': find the duplicate keys and fix the load

#Symptom

A job or streaming query that upserts into a Delta table fails, and every retry fails the same way:

[DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE] Cannot perform Merge as multiple source
rows matched and attempted to modify the same target row in the Delta table in possibly
conflicting ways.

Open-source Delta versions may show the same sentence without the error class. Exact wording varies by version.

Two or more source rows matched one target row, so Delta can't tell which update should win. Retrying won't help because the source data is still duplicated. Fix the source, then rerun.

A failed MERGE commits nothing. To confirm, check that the failed run didn't add a MERGE version:

DESCRIBE HISTORY silver.orders LIMIT 5;

#When the error fires

The last two rows matter most. A job can pass for days while it writes duplicates, then fail when a later batch contains that key twice. Check the target as well as the source.

Note

On Databricks Runtime 15.4 LTS and below, MERGE evaluates only the ON condition when it looks for multiple matches. Newer runtimes also consider WHEN MATCHED conditions, so the same data can fail on an older runtime and pass on a newer one.

#Checks

The examples use staging.orders_batch as the source and silver.orders as the target. Use exactly the same key columns and expressions as your ON clause.

#1. Find the duplicate keys

Run this against the same batch that failed. If the source is a view over "new" data, pin it to the failed window first or the duplicates may have moved on.

SELECT order_id, count(*) AS source_rows
FROM staging.orders_batch
GROUP BY order_id
HAVING count(*) > 1
ORDER BY source_rows DESC
LIMIT 50;

No rows returned? Your ON clause probably isn't a plain equality on those columns. lower(), trim() and casts can make different source values match the same target row, so group by the same expression.

#2. Keep the keys that hit the target

WITH dupes AS (
  SELECT order_id, count(*) AS source_rows
  FROM staging.orders_batch
  GROUP BY order_id
  HAVING count(*) > 1
)
SELECT d.order_id, d.source_rows, count(*) AS target_rows
FROM dupes d
JOIN silver.orders t ON t.order_id = d.order_id
GROUP BY d.order_id, d.source_rows
ORDER BY d.source_rows DESC
LIMIT 50;

#3. Exact copies or conflicting versions?

WITH r AS (
  SELECT order_id, to_json(struct(*)) AS row_json
  FROM staging.orders_batch
)
SELECT order_id,
       count(*) AS source_rows,
       count(DISTINCT row_json) AS distinct_versions
FROM r
GROUP BY order_id
HAVING count(*) > 1
ORDER BY distinct_versions DESC
LIMIT 50;

distinct_versions = 1 means exact copies, usually a replayed file or a double load. A value above 1 means the rows differ. They may be real changes, or differ only in a load timestamp or file name column. Compare a few rows before choosing a fix.

#4. Map it to the cause

Caution

If the ON clause is missing part of the key, deduplicating throws away valid rows without any error. Confirm the table's real business key before you add a dedupe step.

#Fix

#SQL: keep one row per key

Use a column that reflects the real order of changes, such as a source sequence number, log position or source update timestamp. Add a second sort column so ties always resolve the same way.

MERGE INTO silver.orders AS t
USING (
  SELECT * FROM (
    SELECT s.*,
           row_number() OVER (
             PARTITION BY order_id
             ORDER BY source_updated_at DESC, source_seq DESC
           ) AS rn
    FROM staging.orders_batch s
  ) WHERE rn = 1
) AS s
ON t.order_id = s.order_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET
  t.status = s.status,
  t.amount = s.amount,
  t.source_updated_at = s.source_updated_at
WHEN NOT MATCHED AND s.op <> 'D' THEN INSERT
  (order_id, status, amount, source_updated_at)
  VALUES (s.order_id, s.status, s.amount, s.source_updated_at);

List the columns explicitly here. UPDATE SET * would try to include the helper column rn, and fail or write it if schema evolution is on. The op delete handling is for CDC feeds, so remove it if your source has no such column.

Warning

dropDuplicates(["order_id"]) and row_number() without a reliable sort keep an arbitrary row. The load goes green, but the target can end up with an older value. Use them only when the duplicates are exact copies.

#Streaming: dedupe inside each micro-batch

The same key can arrive twice within a micro-batch, so run the dedupe inside foreachBatch:

from delta.tables import DeltaTable
from pyspark.sql import Window, functions as F

def upsert(batch_df, batch_id):
    w = Window.partitionBy("order_id").orderBy(
        F.col("source_updated_at").desc(), F.col("source_seq").desc()
    )
    latest = (batch_df.withColumn("rn", F.row_number().over(w))
              .filter("rn = 1").drop("rn"))

    (DeltaTable.forName(batch_df.sparkSession, "silver.orders").alias("t")
        .merge(latest.alias("s"), "t.order_id = s.order_id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute())

(spark.readStream.table("bronze.orders_cdc")
    .writeStream
    .foreachBatch(upsert)
    .option("checkpointLocation", "abfss://chk@mylake.dfs.core.windows.net/silver_orders")
    .start())

Keep the same checkpoint location when you restart. The failed batch didn't complete, so the query reprocesses it with the new code.

This keeps the latest row within a batch only. To stop a late-arriving batch overwriting newer data, add a condition such as s.source_updated_at >= t.source_updated_at to the update clause.

#Check the target for hidden duplicates

Because duplicate new keys are inserted without an error, check the target before you close the incident:

SELECT order_id, count(*) AS target_rows
FROM silver.orders
GROUP BY order_id
HAVING count(*) > 1
ORDER BY target_rows DESC
LIMIT 50;

If it returns rows, clean them up with your duplicate-record procedure before relying on the table. Note the current version from DESCRIBE HISTORY first so you can restore it if the cleanup goes wrong, and stop writers while it runs.

#Prevent it next time

• Treat duplicate source keys as normal. Replays and CDC bursts will happen, so build the dedupe into every upsert.
• Keep a sequence column from the source. Without one you can't reliably decide which version is newest.
• Don't rely on the table to reject duplicates. Delta doesn't enforce primary keys, and PRIMARY KEY constraints in Unity Catalog are informational.
• Add a uniqueness test on the target key, and alert on it, so inserted duplicates surface before a consumer finds them.
• Test upsert code with a batch that holds the same key twice, once for a new key and once for an existing key.
• Log the number of rows the dedupe removes. A sudden jump usually means a replay upstream or a key that is no longer unique.