Queue Models, Not Commands: A Demand-Driven dbt Scheduler

Two ingestion flows finished within minutes of each other. Both had fresh data and selected the correct downstream dbt models. Then one failed because the two builds had reached the same Delta table.

The SQL was fine. So were the ingests. They had simply arrived at the same table from different sides of the graph.

Delta Lake did what it should. Its optimistic concurrency control validates a write before commit. If another writer commits a conflicting change first, the second operation fails instead of corrupting the table.

The failure was safe. The operating experience was not.

Prefect showed a failed flow, Linear picked up another issue, reporting waited, and I had to prove the model was not broken before restarting it. Retrying worked, but it also made me the scheduler.

Why source-driven dbt builds still collided

The original pattern was simple:

ingest source
    -> dbt build --select source:field_ops.work_orders+

I still wanted that behavior. When an ingest writes new data, its downstream models should follow. When nothing changes, there is no reason to spend compute rebuilding the same outputs.

The problem was the command boundary. Independent ingests could issue different selectors that converged on the same physical model. Serializing whole commands would stop simultaneous writers, but the second command could still rebuild work the first had just finished.

Prefect knew about flows. dbt knew about the graph. What neither gave my independently triggered flows was a shared answer to a smaller question: is this model already waiting or running somewhere else?

That became the job of a small PostgreSQL control plane.

source changed
    -> ask dbt which models are affected
    -> add those model identities to a shared queue
    -> merge work that is already pending or running
    -> let dbt execute one compatible batch

Let dbt plan the work

I did not recreate dbt's graph logic in Python. Before writing anything to the queue, the scheduler asks dbt ls to resolve the selector:

dbt ls \
  --select "source:field_ops.work_orders+" \
  --resource-type model \
  --output json

The trailing + uses dbt's graph selection syntax to include downstream descendants. The JSON output gives the scheduler concrete model IDs and fully qualified selectors without querying the warehouse. The queue stores what dbt resolved, not my interpretation of the DAG.

The planner also applies the same exclusions as the worker. In this execution path, views are excluded because the event-driven job exists to rebuild physical outputs. If a source selector resolves only to views, that is a successful no-op. If it matches no enabled dbt node at all, that remains a planning failure worth investigating.

Queue models, not commands

Suppose two source events resolve to these model sets:

Request A: stg_field_ops__work_orders, dim_project, fact_project_labor
Request B: stg_field_ops__time_entries, dim_project, fact_project_labor

The shared models should not become two builds. They should become one piece of active work with two recorded requests.

The queue handles three common cases:

• No active row: create pending work.
• Already pending: update the request time and merge the selector into the existing row.
• Already running: record the overlap and mark that model for one follow-up pass.

If new source data lands while a model is running, dropping the request could leave the completed model stale. The row therefore carries a requested_while_running flag. On success, a clean row leaves the active queue. A dirty row returns to pending.

The append-only event table records every request. The active row only needs to remember whether it owes another execution, because that execution sees the latest source state.

One batch, one dbt selection

Models share a batch only when their environment, dbt action, scheduler mode, and exclusions match. The worker turns the claimed identities into one selection of fully qualified model names and passes it back to dbt.

dbt still owns dependency order. The queue decides what work is eligible and who owns it. dbt decides how the selection runs.

Failure recovery is part of the queue

A session-level PostgreSQL advisory lock allows one scheduler owner per environment. A second runner exits without claiming work.

Claimed rows carry a batch ID, owner, claim time, and heartbeat. If a worker disappears, the scheduler checks the Prefect run and requeues abandoned rows after a grace period.

Failures are classified before recovery. A Delta metadata or concurrent-write conflict may require rerunning the original selection. Other failures use dbt's retry behavior against failed nodes instead of replaying the whole batch. If recovery still fails, dbt's node-level results show which models completed, so only unfinished rows go back to the queue.

The active queue contains unresolved work. Completed work moves to the event history rather than accumulating forever in the queue table.

Freshness is a policy, not one global schedule

Not every model deserves the same urgency. dim_project may need a short freshness window, fact_project_labor may fit a standard tier, and a historical aggregate may run less often.

Those tiers affect priority, minimum intervals, and how long lower-priority work waits. They change when eligible work runs, not dbt's graph.

The numbers made the design obvious

The event history tracks new_pending, already_pending, and already_running separately. I originally thought of the last two as duplicates. They turned out to be one of the most useful workload signals in the system.

In one recent 24-hour production sample, 55,587 model request events targeted work already pending, and another 44,695 targeted work already running. Only 2,210 request events created new pending rows.

In other words, the scheduler did not merely prevent an occasional write conflict. Most source-driven demand in that sample overlapped with work the platform already knew about.

The financial result pointed in the same direction. In a month-over-month observation after rollout, average daily warehouse cost was about 20% lower and the projected monthly pace was roughly $770 below the prior month. That was not a controlled experiment, so I would not assign every dollar to this scheduler. Thread changes, workload mix, and model performance also mattered.

What I can say confidently is that the queue stopped a large amount of known duplicate demand from becoming duplicate execution. It also meant fewer predictable write races reached Linear and fewer reporting refreshes waited for me to triage them.

Would I build this now that dbt State exists?

I would evaluate dbt State first. It can reuse, clone, or build selected nodes based on logic and data freshness. It currently works with dbt Core through the preview dbt-state plugin on supported versions and warehouses.

That overlaps with this problem, but it does not erase the control plane. I still need to turn source events into demand, merge independent requests, apply freshness policy, recover workers, and preserve operational history.

I would test which pieces dbt State could remove. I would not add Kafka, Kubernetes, or a general-purpose orchestration framework without workload evidence that PostgreSQL and Prefect had stopped being enough.

The broader lesson

The part I care about most is not the queue table or the lock. It is that the ingest no longer throws away what it knows.

The producer knows whether source data changed. dbt knows which models depend on it. PostgreSQL remembers whether that work is already in flight. Each tool keeps the responsibility it is good at, and I am no longer the coordination layer between them.

That is the connection to engineering to zero: remove the condition creating the incident instead of getting faster at retrying it.

Sources and further reading

• About the dbt ls command
• dbt graph operators
• About dbt State
• Setting up dbt State
• Delta Lake concurrency control
• PostgreSQL advisory locks