Delta stream fails with 'Detected a data update in the source table': skipChangeCommits or Change Data Feed?

#Symptom

A Structured Streaming query that reads a Delta table stops with an error like this:

com.databricks.sql.transaction.tahoe.DeltaUnsupportedOperationException:
[DELTA_SOURCE_TABLE_IGNORE_CHANGES] Detected a data update (for example MERGE)
in the source table at version 482. This is currently not supported.
If you'd like to ignore updates, set the option 'skipChangeCommits' to 'true'.

Or, for deletes, you get a similar message with DELTA_SOURCE_IGNORE_DELETE.

It's often a job that ran fine for months, until someone ran a one-off fix or a GDPR delete on the source table.

#Why it happens

A Delta streaming source expects an append-only table. New files mean new rows. If a commit rewrites existing files, as MERGE, UPDATE and DELETE do, Spark can't tell which rows changed. Reading the rewritten files would re-send many unchanged rows, so the stream stops rather than guess.

Not every operation triggers it:

#Step 1: find the commit

The error names the version. Look at it:

DESCRIBE HISTORY main.sales.orders_bronze LIMIT 50;

Focus on the version in the error, and any others like it:

SELECT version, timestamp, userName, operation, operationParameters, operationMetrics, job, notebook
FROM (DESCRIBE HISTORY main.sales.orders_bronze)
WHERE operation IN ('MERGE', 'UPDATE', 'DELETE', 'WRITE', 'RESTORE')
  AND version >= 482
ORDER BY version;

Look at these things:

• Who ran it. Is this a planned job, or a one-off?
• Rows changed. numUpdatedRows, numDeletedRows or numTargetRowsUpdated.
• Will it happen again? A nightly MERGE needs a different answer from a single fix.

#Step 2: decide what downstream must see

This is the real decision. Ask the owner of the target table:

Warning

Choosing skipChangeCommits to get the stream green means the target misses those changes. If a compliance delete ran on the source, the deleted person's data is still in the target. You'll have to apply the delete there as well.

#Option A: skip commits that change data

df = (spark.readStream
      .option("skipChangeCommits", "true")
      .table("main.sales.orders_bronze"))

This ignores the entire commit of any transaction that changes existing rows. Pay attention to what "entire" means here: if a MERGE both updated rows and inserted new ones, those new rows are skipped too.

So if your source gets a MERGE regularly, and those merges add new rows, this option silently loses them. It's only safe when change commits contain nothing that your stream needs.

Adding an option with an existing checkpoint usually works, but test this on a copy first, and read the release notes for your runtime.

Older options still appear in old code:

#Option B: read the Change Data Feed

To apply updates and deletes downstream, read row-level changes.

First, turn it on. It records changes only from this point on, not for earlier history:

ALTER TABLE main.sales.orders_bronze
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

DESCRIBE HISTORY main.sales.orders_bronze LIMIT 1;

Note the version that the property change creates. Then read from the following version:

changes = (spark.readStream
           .option("readChangeFeed", "true")
           .option("startingVersion", 483)
           .table("main.sales.orders_bronze"))

Changing from a normal read to a change feed read needs a new checkpoint. See checkpoint incompatible after a code change.

Each row now has _change_type (insert, update_preimage, update_postimage or delete), _commit_version and _commit_timestamp. Apply these with foreachBatch:

from delta.tables import DeltaTable
from pyspark.sql import functions as F
from pyspark.sql.window import Window

def apply_changes(batch_df, batch_id):
    w = Window.partitionBy("order_id").orderBy(F.desc("_commit_version"), F.desc("_commit_timestamp"))

    latest = (batch_df
              .filter("_change_type != 'update_preimage'")
              .withColumn("rn", F.row_number().over(w))
              .filter("rn = 1")
              .drop("rn"))

    target = DeltaTable.forName(spark, "main.sales.orders_silver")
    (target.alias("t")
        .merge(latest.alias("s"), "t.order_id = s.order_id")
        .whenMatchedDelete(condition="s._change_type = 'delete'")
        .whenMatchedUpdate(
            condition="s._commit_version >= t.source_version",
            set={"status": "s.status", "amount": "s.amount", "source_version": "s._commit_version"})
        .whenNotMatchedInsert(
            condition="s._change_type != 'delete'",
            values={"order_id": "s.order_id", "status": "s.status", "amount": "s.amount", "source_version": "s._commit_version"})
        .execute())

(changes.writeStream
    .foreachBatch(apply_changes)
    .option("checkpointLocation", "/Volumes/main/ops/checkpoints/orders_silver_cdf")
    .trigger(availableNow=True)
    .start())

This uses a source_version column on the target, so a replayed batch can't overwrite newer data. See foreachBatch MERGE idempotency. If you use Lakeflow, you can let AUTO CDC handle the ordering. Read change data feed for more about the feature, including its storage costs and retention. Change files are removed by VACUUM with the rest.

#Option C: rebuild from a new checkpoint

If the source was rewritten, or the target can simply be rebuilt, this is the cleanest route:

# Fix a version first, so that no commit slips between the two steps
version = spark.sql("DESCRIBE HISTORY main.sales.orders_bronze LIMIT 1").first()["version"]

# Replace the target's data with the source at that version, in one batch
(spark.read.option("versionAsOf", version)
   .table("main.sales.orders_bronze")
   .write.mode("overwrite")
   .saveAsTable("main.sales.orders_silver"))

# Continue from just after that version, on a new checkpoint
(spark.readStream
   .option("startingVersion", version + 1)
   .table("main.sales.orders_bronze")
   .writeStream
   .option("checkpointLocation", "/Volumes/main/ops/checkpoints/orders_silver_v2")
   .toTable("main.sales.orders_silver"))

Stop the old stream first, so that two writers don't change the target at once.

Caution

Overwriting a target that holds data no longer in the source, because of retention or VACUUM, loses that data for good. Clone the target first. See deep and shallow clones. Never delete an old checkpoint until the new stream is proven.

#Option D: restructure the source

If people keep changing a bronze table by hand, the design may be the issue:

• Treat bronze tables as append-only. Make corrections in silver.
• For GDPR, design the deletion path upfront across every layer.
• Use table constraints to reject bad data, so it needs less fixing later.

#After the fix

for q in spark.streams.active:
    print(q.name, q.status)
    print(q.lastProgress)

Check numInputRows and the source's startOffset and endOffset, and that row counts agree with the source:

SELECT count(*), max(updated_at) FROM main.sales.orders_silver;
SELECT count(*), max(updated_at) FROM main.sales.orders_bronze;

#Prevent it next time

• Document which tables are append-only streaming sources, and restrict MODIFY on them.
• Alert on stream failures. See job alerts.
• Enable Change Data Feed when you create a table if you expect updates downstream, since it doesn't cover the past.
• Review the source's table history before running any one-off fix.