foreachBatch MERGE in Azure Databricks: duplicates after restarts, 'multiple source rows matched' and merges that get slower every day

#Symptoms

• After a job failure and restart, the target table has duplicates or double-counted values.
• Batches fail with 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").
• Micro-batch duration grows from seconds to minutes as the target table grows.

The standard pattern looks like this:

def upsert(batch_df, batch_id):
    (DeltaTable.forName(spark, "main.sales.orders").alias("t")
       .merge(batch_df.alias("s"), "t.order_id = s.order_id")
       .whenMatchedUpdateAll()
       .whenNotMatchedInsertAll()
       .execute())

(spark.readStream.table("main.raw.orders_cdc")
   .writeStream.foreachBatch(upsert)
   .option("checkpointLocation", "/Volumes/main/chk/orders_merge")
   .trigger(availableNow=True)
   .start())

It works in tests and breaks in production for three separate reasons.

#Problem 1: duplicates after a restart

foreachBatch gives at-least-once semantics. If the job fails after the MERGE committed but before the checkpoint recorded the batch, the same batch_id runs again on restart.

• A MERGE that only does "update all / insert" is naturally idempotent for the same input: running it twice gives the same result.
• It is not idempotent if the batch does increments (t.total = t.total + s.amount), inserts without a match condition, appends to another table, or calls an external system.

Fixes:

Delta idempotent writes for append/overwrite inside foreachBatch:

app_id = "orders_merge_v1"       # stable per query
def upsert(batch_df, batch_id):
    (batch_df.write.format("delta")
        .option("txnAppId", app_id).option("txnVersion", batch_id)
        .mode("append").saveAsTable("main.sales.orders_audit"))

Delta skips a write whose (txnAppId, txnVersion) it has already committed. For MERGE, recent runtimes support the same idea through session settings (spark.databricks.delta.write.txnAppId / txnVersion); check the "idempotent table writes in foreachBatch" docs for your runtime before relying on it.

Make the logic idempotent: write absolute values, not increments; carry a version/sequence column and only update when the source is newer:

.whenMatchedUpdateAll(condition="s.updated_at > t.updated_at")

#Problem 2: multiple source rows match one target row

A micro-batch can contain several changes for the same key. MERGE refuses because the result would be ambiguous.

Deduplicate inside the batch, keeping the latest change per key:

from pyspark.sql import Window, functions as F

def upsert(batch_df, batch_id):
    w = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc(), F.col("cdc_lsn").desc())
    latest = (batch_df.withColumn("rn", F.row_number().over(w))
                      .filter("rn = 1").drop("rn"))
    (DeltaTable.forName(spark, "main.sales.orders").alias("t")
       .merge(latest.alias("s"), "t.order_id = s.order_id")
       .whenMatchedDelete(condition="s.op = 'D'")
       .whenMatchedUpdateAll(condition="s.updated_at >= t.updated_at")
       .whenNotMatchedInsertAll(condition="s.op != 'D'")
       .execute())

Order by a reliable sequence (CDC log sequence number) when timestamps can tie. Don't use dropDuplicates without an order: it keeps an arbitrary row.

#Problem 3: MERGE gets slower every day

MERGE must find matching target rows. Without help it scans the whole target table every batch.

#Give it pruning

If changes only affect recent data, add a target-side predicate:

.merge(latest.alias("s"),
       "t.order_id = s.order_id AND t.order_date >= current_date() - INTERVAL 30 DAYS")

Only do this if it's logically true; a late update for an older row would be inserted as a new row.

#Cluster the target on the merge key

ALTER TABLE main.sales.orders CLUSTER BY (order_id);
OPTIMIZE main.sales.orders;

Liquid clustering co-locates keys, so MERGE reads far fewer files. With predictive optimization enabled on Unity Catalog managed tables, OPTIMIZE runs automatically.

#Deletion vectors

With deletion vectors enabled (default for new tables in recent runtimes), updates mark rows deleted instead of rewriting whole files, which makes small MERGEs much cheaper.

DESCRIBE DETAIL main.sales.orders;           -- check tableFeatures for deletionVectors

#Check what MERGE actually did

DESCRIBE HISTORY main.sales.orders LIMIT 5;

operationMetrics shows numTargetFilesScanned (older runtimes) or similar scan metrics, numTargetRowsUpdated, numTargetRowsInserted and executionTimeMs. Files scanned rising every day = no pruning.

#Problem 4: conflicts with maintenance jobs

A separate OPTIMIZE or DELETE job on the same table can make the streaming MERGE fail with ConcurrentAppendException / ConcurrentDeleteReadException. Row-level concurrency (on tables with deletion vectors and liquid clustering) reduces this; otherwise schedule maintenance between streaming runs or rely on predictive optimization.

#Checklist

Warning

Changing the query logic in a way that changes state or sink semantics may require a new checkpoint. A new checkpoint replays from wherever the source starts; make sure the MERGE is idempotent before doing it.

#Prevent it next time

• Test every foreachBatch by killing the job mid-batch and restarting; compare row counts and totals.
• Always dedupe within the batch for CDC sources.
• Cluster MERGE targets on the merge key from the start.
• Monitor batch duration and MERGE scan metrics, not just success/failure.