Sev1: Delta MERGE runs for hours and the daily SLA is missed — pruning, file layout and quick wins

#Symptoms

• The upsert step in the nightly job runs far longer than usual; downstream reports are late.
• Spark UI shows a huge number of files scanned and a large shuffle for the MERGE.

#1. Look at what the MERGE actually did

DESCRIBE HISTORY prod.sales.orders LIMIT 5;

For the MERGE row, operationMetrics shows numTargetFilesAdded, numTargetFilesRemoved, numTargetRowsUpdated, numTargetRowsCopied and scan/rewrite times. A huge numTargetRowsCopied compared to rows updated means the merge rewrote many files to change a few rows.

DESCRIBE DETAIL prod.sales.orders;   -- numFiles, sizeInBytes, partitionColumns, clusteringColumns

#2. Common causes

#3. Add pruning predicates

If the source only contains recent data, tell Delta:

MERGE INTO prod.sales.orders AS t
USING (SELECT * FROM staging.orders_delta) AS s
ON  t.order_id = s.order_id
AND t.order_date >= date_sub(current_date(), 7)   -- prune target files
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Warning

The extra predicate must be true for every row that can match. If an update arrives for an order older than 7 days, it will be inserted as a duplicate instead of updating. Base the window on the actual source data range, for example the min date in the batch.

A safer dynamic version:

min_date = spark.table("staging.orders_delta").agg({"order_date": "min"}).first()[0]
spark.sql(f"""
MERGE INTO prod.sales.orders t USING staging.orders_delta s
ON t.order_id = s.order_id AND t.order_date >= '{min_date}'
WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *""")

#4. Deduplicate the source

from pyspark.sql import Window, functions as F
w = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
src = (spark.table("staging.orders_delta")
       .withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn"))
src.createOrReplaceTempView("orders_src")

#5. Fix layout

• Compact small files and co-locate keys:

OPTIMIZE prod.sales.orders ZORDER BY (order_id);

• Or, on newer runtimes, switch to liquid clustering on the merge key and date:

ALTER TABLE prod.sales.orders CLUSTER BY (order_date, order_id);
OPTIMIZE prod.sales.orders;

• Turn on deletion vectors so updates don't rewrite whole files:

ALTER TABLE prod.sales.orders SET TBLPROPERTIES ('delta.enableDeletionVectors' = true);

Check that every reader of the table supports the resulting protocol before enabling table features.

#6. Get today's run done

• Kill the running merge if it is far from finishing.
• Apply the predicate and dedup fix (code change only, quick to review).
• Run on a larger job cluster for this run with Photon enabled, if your policy allows.
• Run OPTIMIZE after the load, not before, if time is short.

#Prevent it next time

• Always include a partition/date predicate derived from the source batch.
• Schedule OPTIMIZE (or predictive optimization for Unity Catalog managed tables).
• Track merge duration and files rewritten per run; alert on a jump.
• Reject source batches with duplicate keys in a data quality step.