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.

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.

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.

Module 4: Tables: Parquet, Delta Lake and Iceberg

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

Take this module into your AI

ChatGPT Claude

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 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:

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()

+-----------------+----------------+----------------+-----------------+
|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:

Try itMERGE, row by row

The batch, s

1001 shipped 10-02 08:00the same change again: left aloneSkip
1002 placed 10-01 10:02older than the table's row: left aloneSkip
1004 shipped 10-04 08:30newer than 10-02 12:45: UPDATE SET *Update
1005 cancelled 10-04 10:00newer than 10-03 14:10: UPDATE SET *Update
1006 placed 10-04 11:20no order with this id: INSERT *Insert

Updated2

Inserted1

The table after the MERGE

1001 shipped
1002 cancelled
1004 shipped
1005 cancelled
1006 placed

2 updated, 1 inserted. The condition s.updated_at > t.updated_at lets only a newer change through, so the repeated 1001 and the late, older row for 1002 are left alone. Run the same batch again and nothing changes.

MERGE INTO orders t USING changes s ON t.order_id = s.order_id WHEN MATCHED, with or without AND s.updated_at > t.updated_at, THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *. Run on delta-spark 4.4.1; times are in October 2026.

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:

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()

[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, 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 adds a column to the orders, and shows the change MERGE drops without a word.

Check yourself

2 questions, each in a new situation

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?
Not quite. 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. A batch has 2 rows for order 1007, which is not yet in the table. What does MERGE ... WHEN NOT MATCHED THEN INSERT * do?
Not quite. 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