Backfill Concurrency and Collisions with the Live Schedule

Reprocessing a year of history is a different workload from running today’s batch, but it uses the same tasks, the same workers, and the same warehouse. A backfill that ignores this produces a week of latency on the live schedule, or a corrupted partition where two runs wrote the same target, or both. The collision is not a scheduler bug; it is the ordinary consequence of two instances of the same graph addressing different intervals at the same time.

The shape of the contention

A backfill of n intervals creates n runs, each with the full task graph. If the graph has width w and the scheduler allows c concurrent tasks, the live schedule’s tasks and the backfill’s tasks compete for c slots. With no policy, the competition is decided by arrival order in the ready queue, and a thousand ready backfill tasks will fill every slot, delaying the live run by however long the backfill takes.

The delay is not uniform across the live run. The first tasks of the live run may get slots immediately, but the tasks that become ready later, after a dependency completes, queue behind the entire backfill. So the live run starts on time and then stalls mid-graph, which is harder to diagnose than a run that simply did not start.

Three resources are contended, and they are not the same resource.

Worker slots are the scheduler’s own limit, expressed as a maximum number of concurrent tasks. They bound CPU and memory only if each task’s footprint is similar; a backfill of small metadata tasks and a live run of a large aggregation share the slot count but not the resource.

Warehouse capacity is the query engine’s limit, usually expressed as concurrent query slots or as a cap on consumed compute units. It is invisible to the scheduler, so a scheduler that dispatches twenty tasks against a warehouse that accepts eight produces twelve tasks that wait inside the warehouse’s queue, holding their worker slots while doing nothing.

Storage bandwidth is the limit that appears last and hurts most. A backfill reading the same table that the live run is writing, or writing partitions that a downstream compaction is reading, saturates the object store’s request rate. The symptom is not a queue but a rise in latency across every task touching that storage, including tasks outside the backfill.

Separating the pools

The simplest arbitration is a separate resource pool per class of work. The scheduler’s pool mechanism assigns each task to a named pool with a slot count, and a task cannot run unless its pool has a free slot. Giving backfill tasks their own pool with a small slot count, two or three against the live schedule’s twenty, bounds the backfill’s share of the scheduler’s capacity.

def backfill_slots(warehouse_slots, slots_per_task, live_share=0.8):
    """Slots the backfill may use without starving the live schedule.

    warehouse_slots is the engine's concurrent-query limit, not the scheduler's.
    live_share is the fraction of that limit reserved for scheduled runs.
    """
    reserved = int(warehouse_slots * live_share)
    spare = warehouse_slots - reserved
    return max(0, spare // slots_per_task)

The pool bounds the scheduler’s dispatch, not the warehouse’s acceptance, so the arithmetic above is only as good as its input. slots_per_task is how many of the engine’s concurrent query slots one task occupies, which depends on the query’s shape and on whether the engine counts a slot per statement or per unit of memory. An estimate that is too low over-commits the engine, and the symptom is a queue inside the engine rather than a queue in the scheduler. The pool size must therefore be chosen against the downstream resource rather than against the scheduler’s own capacity, which requires knowing what one task costs at the warehouse, a fact about the query rather than about the scheduler’s configuration.

Where the warehouse charges by compute unit rather than by concurrent slot, the bound has to be on the units, and schedulers generally do not model that. The practical approach is to give the backfill’s tasks a smaller warehouse size or a lower priority tier so that the warehouse’s own scheduler favours the live queries. The trade is that the backfill’s tasks take longer individually, so the pool must be sized for the longer duration to achieve the same throughput.

Partition locks and the same-target problem

Two runs of the same graph for different intervals usually write different partitions, and there is no collision. The exception is the run that writes a target not partitioned by the logical date: a full-refresh table, a slowly changing dimension, an aggregate over all history. Two intervals of a backfill writing such a table concurrently produce an outcome that depends on commit order, and the last writer’s content wins.

The same problem appears within a single interval when the live schedule and a backfill both target the current day. A backfill launched to repair the 24th, while the live run for the 24th is still retrying, gives two runs for the same interval. Most schedulers refuse to create a second run for an interval that already has one, which is the correct default; the override that permits it is exactly the operation that needs a lock.

A partition lock keyed on the target is the mechanism. The lock is acquired before the write and released after the commit, and its granularity is the partition, not the task. Two tasks writing different partitions proceed; two tasks writing the same partition serialise. Implementing the lock in the scheduler’s state store makes it visible and revocable; implementing it in the storage layer, as a conditional commit or a table lock, makes it enforceable even for writers the scheduler does not control.

The cost of a partition lock is that it converts parallelism into waiting, and a backfill with an unpartitioned target becomes serial regardless of its pool size. That is a correct reflection of the underlying constraint: if every interval writes the same table, the intervals cannot be written concurrently without a merge strategy. The merge strategy, accumulating into a staging table per interval and applying the deltas in order, restores parallelism at the cost of an extra stage and an ordering requirement on the apply step.

Rate, not just concurrency

Bounding the number of concurrent tasks bounds the instantaneous load. It does not bound the total work, and a backfill’s total work is what determines how long the live schedule is degraded. A backfill that runs three tasks at a time for eight hours degrades the live run for eight hours, even though no single moment is oversubscribed.

The control that addresses this is a rate or a budget: a maximum number of task completions per minute, or a maximum total compute consumed per hour, shared with the live schedule. Rate limiting is harder to implement than a slot count because it requires a shared counter with a time window, but it expresses the constraint that actually matters for a long-running background workload.

The alternative is to schedule the backfill into a quiet period. A warehouse that is idle between midnight and six, local time, can absorb a backfill that is paused during the hours the live schedule runs. This is crude, it wastes the backfill’s opportunity to use otherwise-idle capacity, and it depends on the live schedule’s timing being stable, but it requires no coordination mechanism at all, which is why it is common.

Detecting the collision after the fact

The collision that matters most is not the one that slows a run but the one that corrupts a partition. Its signature is a target whose contents do not match any single run’s output: a row count between the two runs’ counts, a maximum timestamp later than either run’s interval, or a mixture of values from two different source snapshots.

Detecting it requires the target to carry provenance. A column recording the run that wrote each row, or a manifest listing the files each commit added, makes the mixture visible. Without provenance, the only signal is an inconsistency discovered by a downstream consumer, at which point the interval of the corruption is unknown and the repair must cover a range rather than a point.

The cheaper prevention is to make every write replace its partition atomically, so that a second writer’s commit is a complete replacement rather than an interleaving. With atomic partition replacement, two concurrent writers produce one of the two results, not a mixture, and the failure mode degrades from corruption to a last-writer-wins race that can be detected by comparing the committed run identifier against the expected one. That comparison is a single check at the end of the run, and it is the only verification that the scheduler’s own state cannot provide.

What the live schedule should do

The live schedule has no knowledge of the backfill unless it is told. If the backfill is bounded by a pool and a rate, the live schedule’s latency rises but its correctness is unaffected, and no change to the live graph is needed. If the backfill shares a target with the live run, the live run needs to detect the race, which means the check described above belongs in the live graph rather than only in the backfill.

The asymmetry is worth stating: the backfill is an exceptional operation performed deliberately, so it can be required to carry extra guards. The live schedule runs every day without an operator watching, so any guard it carries must be cheap and must fail in a way that is obvious. Placing the partition-replacement check in the live graph costs one task and turns a silent race into a failed run, the right trade for a job that must be trustworthy unattended.

Backfills also interact with retries in a way that compounds: a backfill’s task that fails and retries occupies its pool slot for the retry duration, so a pool of three with a five-attempt retry policy can be effectively reduced to one for the length of the backoff. Sizing the pool without accounting for retry policy produces a backfill that runs far slower than its slot count suggests, and the natural response, raising the slot count, increases the load on the warehouse that the pool was meant to protect.