# MERGE INTO in Delta Lake: upsert a batch of changes safely

MERGE INTO matches each row of a batch to the table on a key, then updates the rows that match and inserts the rest, in 1 commit. Add a condition such as s.updated_at > t.updated_at to the update, so a change applies only when it is newer than the row in the table: running the same batch again then changes nothing. Keep 1 row per key in the batch first, or Delta stops with DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE for a key in the table, and inserts a new key twice.

- Page: https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/merge
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 4: Tables: Parquet, Delta Lake and Iceberg, MERGE
- Parts of this module: The transaction log and time travel (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg); MERGE (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/merge); Schema enforcement and evolution (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/schema); OPTIMIZE, clustering and VACUUM (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/optimize-and-vacuum); Iceberg and UniForm (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/iceberg-and-uniform); Build it: an incremental load (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/build-it)

At the end of this part you can:

- Upsert a batch of changes into a Delta table with MERGE INTO, in SQL and in Python
- Make a merge safe to run twice, and safe from late, older changes
- Fix the multiple source rows error, and the duplicate inserts it does not catch

The orders table changes every day: orders ship, some are cancelled, new ones arrive. The source system sends the day's changes as a batch, and the table must take them in without losing or doubling an order. That is an upsert: update the orders the table has, insert the ones it does not. Delta does it in 1 statement, `MERGE INTO`.

## Merge a batch of changes

Here is a day's batch, against the 4 orders [part 1](/data-engineering/course/delta-lake-and-iceberg) restored. 1004 ships, 1005 is cancelled and 1006 is new. 2 rows are traps, as real feeds have: 1001 again with nothing changed, and an old version of 1002, sent late, from before it was cancelled. Save this as `merge.py`:

```python
from pyspark.sql import functions as F
from delta_session import delta_spark

spark = delta_spark()
changes = spark.createDataFrame(
    [
        (1001, "C01", "shipped", "25.50", "2026-10-02 08:00:00"),
        (1002, "C02", "placed", "12.00", "2026-10-01 10:02:00"),
        (1004, "C01", "shipped", "40.00", "2026-10-04 08:30:00"),
        (1005, "C09", "cancelled", "8.75", "2026-10-04 10:00:00"),
        (1006, "C02", "placed", "19.90", "2026-10-04 11:20:00"),
    ],
    "order_id BIGINT, customer_id STRING, status STRING, amount STRING, updated_at STRING",
).select(
    "order_id", "customer_id", "status",
    F.col("amount").cast("decimal(10,2)").alias("amount"),
    F.col("updated_at").cast("timestamp").alias("updated_at"),
)
changes.createOrReplaceTempView("changes")

merge = """
MERGE INTO orders AS t
USING changes AS s
ON t.order_id = s.order_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
"""
spark.sql(merge).show()
spark.sql(merge).show()
spark.sql("SELECT order_id, status, updated_at FROM orders ORDER BY order_id").show()
```

```text
+-----------------+----------------+----------------+-----------------+
|num_affected_rows|num_updated_rows|num_deleted_rows|num_inserted_rows|
+-----------------+----------------+----------------+-----------------+
|                3|               2|               0|                1|
+-----------------+----------------+----------------+-----------------+

+-----------------+----------------+----------------+-----------------+
|num_affected_rows|num_updated_rows|num_deleted_rows|num_inserted_rows|
+-----------------+----------------+----------------+-----------------+
|                0|               0|               0|                0|
+-----------------+----------------+----------------+-----------------+

+--------+---------+-------------------+
|order_id|   status|         updated_at|
+--------+---------+-------------------+
|    1001|  shipped|2026-10-02 08:00:00|
|    1002|cancelled|2026-10-03 09:00:00|
|    1004|  shipped|2026-10-04 08:30:00|
|    1005|cancelled|2026-10-04 10:00:00|
|    1006|   placed|2026-10-04 11:20:00|
+--------+---------+-------------------+
```

Read the statement clause by clause. `ON` says how a change finds its order. `WHEN MATCHED` acts on changes to orders the table has, and `UPDATE SET *` copies every column across. `WHEN NOT MATCHED` acts on orders the table has not seen, and `INSERT *` adds them.

The condition `AND s.updated_at > t.updated_at` is what makes this safe. It lets a change through only when it is newer than the row it would replace, so 1001's repeat and 1002's late row are left alone. It also makes the second run change nothing, so a job that fails after its MERGE can simply run again. Take it out and the late row turns 1002 back into a placed order. Try both:

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/merge)

## 2 rows for 1 order

Batches often hold more than 1 change for an order, here 1004 shipped and then delivered. Save this as `merge_twice.py`:

```python
from pyspark.sql import Window
from pyspark.sql import functions as F
from delta import DeltaTable
from delta_session import delta_spark

spark = delta_spark()
changes = spark.createDataFrame(
    [
        (1004, "C01", "delivered", "40.00", "2026-10-05 09:00:00"),
        (1004, "C01", "returned", "40.00", "2026-10-06 17:30:00"),
        (1007, "C03", "placed", "30.00", "2026-10-05 09:00:00"),
        (1007, "C03", "shipped", "30.00", "2026-10-05 15:00:00"),
    ],
    "order_id BIGINT, customer_id STRING, status STRING, amount STRING, updated_at STRING",
).select(
    "order_id", "customer_id", "status",
    F.col("amount").cast("decimal(10,2)").alias("amount"),
    F.col("updated_at").cast("timestamp").alias("updated_at"),
)


def merge(batch):
    return (
        DeltaTable.forName(spark, "orders").alias("t")
        .merge(batch.alias("s"), "t.order_id = s.order_id")
        .whenMatchedUpdateAll(condition="s.updated_at > t.updated_at")
        .whenNotMatchedInsertAll()
        .execute()
    )


try:
    merge(changes)
except Exception as error:
    print(str(error).split("]")[0] + "]")

newest_first = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
latest = changes.withColumn("n", F.row_number().over(newest_first)).filter("n = 1").drop("n")
merge(latest).show()
spark.sql("SELECT order_id, status FROM orders WHERE order_id IN (1004, 1007)").show()
```

```text
[DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE]
+-----------------+----------------+----------------+-----------------+
|num_affected_rows|num_updated_rows|num_deleted_rows|num_inserted_rows|
+-----------------+----------------+----------------+-----------------+
|                2|               1|               0|                1|
+-----------------+----------------+----------------+-----------------+

+--------+--------+
|order_id|  status|
+--------+--------+
|    1004|returned|
|    1007| shipped|
+--------+--------+
```

The first MERGE fails as a whole, and the table is unchanged: 2 rows match order 1004, and Delta will not guess which wins. The same code is the Python form of the statement, the `DeltaTable` API, which suits a job that builds its batch in PySpark.

The error protects only orders already in the table. 2 rows for the new order 1007 match nothing, so without the fix MERGE inserts both, and the table holds 1007 twice with no error at all. The fix covers both: keep the newest row per order with `row_number()`, as in [module 2](/data-engineering/course/pyspark-dataframes/keep-the-newest-row), then merge.

## Deletes

MERGE can also delete. `WHEN MATCHED AND s.status = 'deleted' THEN DELETE` removes orders the source flags as deleted. `WHEN NOT MATCHED BY SOURCE THEN DELETE` removes rows of the table that the batch does not have, which is right only when the source sends the full table every time; with a batch of changes, it would delete every order that did not change that day.

On Databricks, the SQL and the `DeltaTable` code are the same, with the table name from your catalog and schema. [Part 3](/data-engineering/course/delta-lake-and-iceberg/schema) adds a column to the orders, and shows the change MERGE drops without a word.

## Check yourself

1. Your MERGE uses WHEN MATCHED THEN UPDATE SET * with no condition. Yesterday's batch is loaded again by mistake, after today's. What happens to an order that changed in both?

   A. Nothing, because MERGE skips rows it has seen before
   B. The MERGE fails, because the order already exists
   C. The order goes back to yesterday's state, because every matched row is overwritten with the batch's values
   D. Delta keeps the row with the later commit time

2. A batch has 2 rows for order 1007, which is not yet in the table. What does MERGE ... WHEN NOT MATCHED THEN INSERT * do?

   A. It fails with DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE
   B. It inserts both rows, so the table has order 1007 twice
   C. It inserts the row with the later updated_at
   D. It inserts the first row and skips the second

### Answers

1. C. Without a condition, WHEN MATCHED updates every row that matches, whatever its values. The older batch overwrites today's changes. A condition such as AND s.updated_at > t.updated_at applies a change only when it is newer than the row in the table, so an old or repeated batch updates nothing.
2. B. The multiple matches error is about rows that match a row already in the table. 2 batch rows with a new key match nothing, so each is inserted, and the table now holds 2 rows for 1 order with no error. Keeping only the newest row per key in the batch, before the MERGE, prevents both problems.

## Questions

### How do I upsert into a Delta table?

Use MERGE INTO target USING source ON the key, then WHEN MATCHED THEN UPDATE SET * and WHEN NOT MATCHED THEN INSERT *. In Python, DeltaTable.forName(spark, name).alias("t").merge(source.alias("s"), condition) builds the same statement, ending with execute(). Both run as 1 commit and return the numbers of rows updated, inserted and deleted.

### What does DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE mean?

2 or more rows of the source matched the same row of the target, so Delta cannot tell which should update it, and the whole MERGE fails with no change to the table. Keep 1 row per key in the source before the MERGE, usually the newest change, with row_number over a window partitioned by the key and ordered by the change time, newest first.

### How do I make a Delta MERGE idempotent?

Make the update conditional on the change being newer, as in WHEN MATCHED AND s.updated_at > t.updated_at THEN UPDATE SET *, and keep 1 row per key in the batch. A second run of the same batch then finds no newer rows and nothing new to insert, so it changes no rows. It still writes a commit, with 0 rows changed.

### Can MERGE delete rows?

Yes. WHEN MATCHED AND a condition THEN DELETE removes matched target rows, for example changes flagged as deletes, and WHEN NOT MATCHED BY SOURCE THEN DELETE removes target rows that no source row matches, which suits a source that is a full copy of the table rather than a batch of changes.

## Sources

- [Table deletes, updates, and merges](https://docs.delta.io/latest/delta-update.html)
- [Upsert into a Delta Lake table using merge](https://docs.databricks.com/aws/en/delta/merge)
- [pyspark.sql.functions.row_number](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.row_number.html)