How do I upsert (MERGE) into a Delta table?

Summary

• Upsert into a Delta table with MERGE INTO in SQL or DeltaTable.merge() in Python, Scala or Java: join a source to the target on a key, update the rows that match, and insert the rest.
• Delta Lake commits all of a merge's changes as one new table version, so readers keep a consistent snapshot while it runs.
• If several source rows match the same target row and the merge tries to update it, the merge fails. Reduce the source to the latest change per key first.
• For streams, run the merge inside foreachBatch and keep it idempotent, because a restarted query can apply the same micro-batch again.

How do I upsert (MERGE) into a Delta table?

Run MERGE INTO (or DeltaTable.merge()) with the target table, a source of new records, and a match condition on the key. WHEN MATCHED updates rows that already exist and WHEN NOT MATCHED inserts new ones. Delta Lake stages the rewritten files and commits them as one new table version, so readers never see a half-applied upsert.

The SQL form

In open-source Apache Spark, Delta SQL commands need a session configured with spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension and spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog. This statement also deduplicates the source inline, keeping the newest change for each id:

MERGE INTO customers AS t
USING (
  SELECT id, name, email, updated_at FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) AS rn
    FROM customer_updates
  ) ranked WHERE rn = 1
) AS s
ON t.id = s.id
WHEN MATCHED AND s.updated_at > t.updated_at THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

The condition on WHEN MATCHED skips stale changes: when no matched clause applies, Delta leaves the target row unchanged.

The Python form

The delta-spark package exposes the same operation as a builder:

from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "/data/customers")

(target.alias("t")
    .merge(updates_df.alias("s"), "t.id = s.id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

Use whenMatchedUpdate(set={...}) and whenNotMatchedInsert(values={...}) to map columns explicitly.

Clause rules to know

Clause	What it does	Rule to remember
WHEN MATCHED	Updates or deletes the matched target row	Can have at most one update and one delete action. With several clauses, every clause except the last needs a condition.
WHEN NOT MATCHED	Inserts a source row that has no match	Insert only. Target columns you don't list are set to NULL.
WHEN NOT MATCHED BY SOURCE	Updates or deletes target rows missing from the source	Delta Lake 2.3 and above for the APIs, 2.4 and above for SQL. Add a condition so the whole target isn't rewritten.
UPDATE SET * / INSERT *	Maps all columns by name	Without schema evolution, source columns the target lacks are ignored.

Avoid the two common failures

Multiple matches. Two source rows matching one target row make the update ambiguous under SQL merge semantics, so Delta fails with DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE. Fix the source: keep only the latest change per key, as the ROW_NUMBER() subquery above does.
Schema drift. A merge doesn't add columns by default. To add new source columns to the target, call .withSchemaEvolution() on the merge builder (Delta Lake 3.2 and above) or set spark.databricks.delta.schema.autoMerge.enabled to true before the merge.

Upserts from a stream

For a stream, call the merge from foreachBatch, which passes your function each micro-batch as a DataFrame plus its batch ID:

def upsert_batch(batch_df, batch_id):
    (target.alias("t")
        .merge(batch_df.alias("s"), "t.id = s.id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute())

(updates_stream.writeStream
    .foreachBatch(upsert_batch)
    .option("checkpointLocation", "/checkpoints/customers")
    .start())

foreachBatch gives at-least-once write guarantees by default, and a restart can replay a batch. Keep the merge idempotent: an upsert that sets columns to source values gives the same result when repeated, and a running total such as t.total = t.total + s.amount double-counts.

Tune merge performance

• Narrow the search. By default, merge searches the entire target for matches. Add known constraints, such as a partition or date range, to the ON condition.
• Compact small files, and tune spark.sql.shuffle.partitions, which sets the merge's shuffle parallelism and the number of output files.
• Repartition before write. On partitioned tables, where each shuffle task can write files into many partitions, setting spark.databricks.delta.merge.repartitionBeforeWrite.enabled to true repartitions output by the partition columns first.
• Enable deletion vectors. From Delta Lake 3.1.0, MERGE on a table with delta.enableDeletionVectors set to true marks changed rows instead of rewriting whole Parquet files. Enabling them upgrades the protocol.
• Keep concurrent merges disjoint. Under optimistic concurrency, writers that touch the same files can conflict, and one fails. Partitioning by the columns your conditions use keeps their file sets apart.

How Databricks approaches MERGE

Databricks runs the same MERGE INTO and DeltaTable code. Its docs also cover:

• Schema evolution in SQL. MERGE WITH SCHEMA EVOLUTION INTO enables schema evolution for one merge, and Databricks recommends per-write schema evolution over a Spark configuration. See schema evolution.
• Row-level concurrency. Row-level conflict detection reduces write conflicts between concurrent operations on the same data files. See isolation levels and write conflicts.
• CDC without hand-written merges. In Lakeflow Pipelines, AUTO CDC (AUTO CDC ... INTO in SQL, create_auto_cdc_flow() in Python) applies a change feed as SCD Type 1 or Type 2 and handles sequencing. See AUTO CDC. Open-source Spark Declarative Pipelines gained SCD Type 1 Auto CDC in Apache Spark 4.2.

FAQs

Can MERGE delete rows?

Yes. WHEN MATCHED ... THEN DELETE (whenMatchedDelete() in Python) removes matched rows, which is how you apply delete flags from a change feed. WHEN NOT MATCHED BY SOURCE ... THEN DELETE removes target rows that the source no longer contains.

Why can a merge produce wrong rows when the source calls current_timestamp()?

Merge usually reads the source twice, and a nondeterministic expression can differ between passes. Delta Lake 2.2 and above materializes the source automatically to prevent this. On older versions, save the source to a temporary Delta table first.

Is an insert-only MERGE useful?

Yes. A merge with only WHEN NOT MATCHED THEN INSERT * skips records that already exist, which deduplicates appends. Because it only appends, other queries can still read the table as a stream.