Idempotent Backfills for Batch Pipelines: Partition Overwrite, Run Manifests and Reprocessing Without Double Counting

Why backfills double count

A batch job that appends is not idempotent, and a backfill is the moment that matters. Someone fixes a bug in a daily aggregate and reruns the last thirty days; every day that already had rows now has them twice. A job that failed after writing half a partition and is retried writes the other half and the first half again. A job whose query filters on the current time rather than on the interval it was asked to process produces different rows on every rerun. None of these are exotic; they are the default behaviour of a pipeline that was written for the first run and never for the second.

The fix has one principle: a run owns exactly one data interval, reads only inputs for that interval, and replaces exactly the output partitions for that interval. Everything else in this article is a way of making that principle hold in Spark, Iceberg and Airflow, and a way of noticing when it does not.

Bound every job by its data interval

Airflow gives each DAG run a data interval, and the DAG run documentation is explicit about two things that break idempotency. First, catchup: with catchup set to true, the scheduler creates a run for every data interval that has not been run since the last one, so a DAG that is not written to handle its interval, one that queries up to now instead of up to the interval end, must have catchup off. Second, manually triggered runs: the documentation warns not to assume that the run data interval is derived from or equal to the supplied logical date, and to use the interval fields explicitly instead of the logical date. The practical rule is that the query takes data_interval_start and data_interval_end as parameters and uses nothing else to decide which rows it processes.

Replace, do not append: partition overwrite modes

Spark has two partition overwrite modes, documented under spark.sql.sources.partitionOverwriteMode. In static mode, the default, Spark deletes all partitions matching the PARTITION specification in the INSERT statement before overwriting. In dynamic mode Spark does not delete partitions ahead of time and overwrites only the partitions that have data written into them at runtime. The setting can also be passed as a write option named partitionOverwriteMode, which takes precedence over the session setting.

Iceberg adds two facts. Overwrites are atomic operations for Iceberg tables, so a reader never sees a half-replaced partition. And the Iceberg Spark writes page recommends dynamic mode when writing to Iceberg tables, while recommending MERGE INTO over INSERT OVERWRITE altogether, because MERGE can rewrite only the affected data files and because the data replaced by a dynamic overwrite may change if the partitioning of the table changes. The same page notes that static mode cannot replace hidden partitions such as hours derived from a timestamp, because the PARTITION clause can only reference table columns.

Illustrative sql, checked for syntax; adapt names and storage to your system.

-- Dynamic overwrite: only partitions with rows in the SELECT result are replaced.
SET spark.sql.sources.partitionOverwriteMode = dynamic;

INSERT OVERWRITE prod.analytics.daily_orders
SELECT CAST(order_ts AS DATE) AS day, region, COUNT(*) AS orders, SUM(amount) AS revenue
FROM prod.raw.orders
WHERE order_ts >= TIMESTAMP '2026-09-18 00:00:00' AND order_ts < TIMESTAMP '2026-09-19 00:00:00'
GROUP BY CAST(order_ts AS DATE), region;

Illustrative PySpark; the option name and semantics are from the Spark configuration reference.

def write_interval(spark, df, table, start, end):
    """Replace exactly the partitions produced for [start, end). df must already be filtered to that interval."""
    (df.write.format("iceberg")
       .mode("overwrite")
       .option("partitionOverwriteMode", "dynamic")
       .saveAsTable(table))
    return {"table": table, "start": str(start), "end": str(end), "rows": df.count()}

The empty-result trap and the run manifest

Dynamic overwrite has one behaviour that bites during backfills. If a rerun produces no rows for a partition, because an upstream source was missing or a filter was wrong, dynamic mode writes nothing and leaves the old partition exactly as it was. The job succeeds, the data is stale, and nothing says so. Static mode with an explicit PARTITION clause would have deleted the partition, which is at least visible; the documentation notes that static mode with the clause omitted replaces all partitions, which is the opposite disaster.

The defence is a run manifest: a small table with one row per job and interval recording the input snapshot or file list, the output row count, the output partitions touched and the run status. Before a backfill commits, compare its row count with the previous run for the same interval and with a count from the source; a drop to zero, or a change beyond a tolerance, fails the run before publication. When an interval legitimately has no data, the job deletes the partition explicitly and records zero in the manifest, so that an empty partition is a decision rather than an accident.

Illustrative python, checked for syntax; adapt names and storage to your system.

def check_against_manifest(manifest, job, start, end, new_rows, source_rows, tolerance=0.02):
    """manifest: {(job, start, end): {"rows": int, "status": str}} from the manifest table."""
    previous = manifest.get((job, start, end))
    problems = []
    if source_rows == 0 and new_rows == 0:
        return ["interval has no source data: delete the partition explicitly and record zero"]
    if new_rows == 0:
        problems.append("rerun produced zero rows while the source has data")
    if previous and previous["status"] == "published" and previous["rows"]:
        change = abs(new_rows - previous["rows"]) / previous["rows"]
        if change > tolerance:
            problems.append(f"row count moved {change:.1%} against the published run")
    if source_rows and abs(new_rows - source_rows) / source_rows > tolerance:
        problems.append(f"output {new_rows} does not reconcile with source {source_rows}")
    return problems

Write-audit-publish with Iceberg branches

For tables where a bad backfill is expensive, Iceberg branches let the job write somewhere readers do not look. The branching documentation describes the audit branch pattern: set the table property write.wap.enabled to true, create a branch with its own retention, write to it, validate, and only then publish. The Spark writes page adds that branch writes can be performed as part of a write-audit-publish workflow by specifying the spark.wap.branch configuration, with the note that the WAP branch and an explicit branch identifier cannot both be specified. The manifest checks from the previous section run against the audit branch, and publication becomes a metadata operation on the main branch rather than a second copy of the data. Branch retention matters: the documentation gives examples of branches retained for seven days with a bounded number of snapshots, which is enough to inspect a failed backfill without keeping its files forever.

Running the backfill: Airflow 3 controls

Airflow 3 exposes backfills as first-class objects. The DAG run documentation shows the CLI form: airflow backfill create with a DAG id, from and to dates, a reprocess behaviour, a maximum number of active runs, an optional backwards ordering and a run configuration; the UI form offers the same fields. Two of those controls carry the idempotency story. Reprocess behaviour set to failed reruns only the intervals that failed, which is the right default when the job is idempotent per interval. Max active runs protects the warehouse from thirty concurrent overwrites, and running backwards produces the most recent, most looked-at days first. The documentation also describes clearing task instances to rerun them, which increments the try number and preserves history; that is the right tool for one bad interval, while backfill is for a range.

Worked example: thirty-one days, one failed day

A daily aggregate for September 2026 is rerun after a fix. The backfill is created for 1 to 30 September with reprocess behaviour set to failed, three active runs and backwards ordering. Twenty-nine intervals had succeeded before and are not rerun; the one that failed is. Each rerun filters orders to its interval, writes with dynamic overwrite to an audit branch, and runs the manifest check: the new row count for 18 September is 41,207 against 41,190 published and 41,207 in the source, a 0.04 percent move inside tolerance, so the branch is published. A second rerun of the same day, triggered by mistake, replaces the same partition with the same rows and records a second manifest line with the same count. The figures are illustrative; the shape is what to copy.

Failure modes

• Filters on now(). Every rerun processes a different window. Parameterise on the data interval only.
• Append mode on retry. A task that appended half a partition and retried appends again. Use overwrite per interval or MERGE on a key.
• Static overwrite without a PARTITION clause. The documentation is explicit: all partitions are replaced. One typo rewrites the table.
• Partition spec evolution. After changing the partitioning, a dynamic overwrite replaces different files than before, which the Iceberg documentation gives as a reason to prefer MERGE INTO.
• Late data outside the interval. Rows that arrive after their interval was published need a separate late-arrival run that targets the original partition, not the current one; see watermarks in stream processing for the streaming equivalent.
• Concurrent writers on one partition. A scheduled run and a backfill touching the same interval race; Iceberg fails one commit on conflict, so the job must treat a commit conflict as a retryable error, not a success.
• Non-deterministic columns. A current_timestamp() column in the output makes every rerun a change, defeating the manifest comparison.

Practice and related reading

Further reading:

• Watermarks in stream processing
• The Iceberg and Trino stack
• Change data capture in 2026
• Idempotency in system design
• Flink checkpoint design

Sources and scope

Checked against these sources on 18 October 2026. Examples and figures are illustrative, not measurements.

• Apache Spark configuration: spark.sql.sources.partitionOverwriteMode
• Apache Iceberg documentation: Spark writes, INSERT OVERWRITE and WAP branch writes
• Apache Iceberg documentation: branching and tagging, audit branches
• Apache Airflow documentation: DAG runs, data intervals, catchup and backfill