Runs, Task States, and the Data Interval
Read an orchestrator's own metadata to tell which runs and tasks are healthy, and explain why a run is responsible for a time range rather than for the moment it happened to execute.
A run owns a time range, not a moment
An orchestrator stores two things about every execution, and the interview answer lives in the difference between them.
A DAG run is one execution of the whole graph. It is bound to an immutable data interval: a half-open time range [data_interval_start, data_interval_end) that the run is responsible for. A daily daily_sales run for 2026-03-13 owns everything that happened between 2026-03-13 00:00 and 2026-03-14 00:00. That is the run's identity. It does not change if the run is retried at midnight, re-run by hand a week later, or backfilled next quarter.
Separately, the run has a wall clock: started_at and ended_at, when the machine actually did the work. A scheduled daily run starts just after its interval closes, so the two look almost the same and beginners conclude they are the same thing. A backfill proves they are not. A backfill run for 2026-02-01 that you launch on 2026-03-16 has a data_interval_end six weeks before its started_at.
The whole reason this matters is that your pipeline logic must filter on the interval, never on the clock. A job whose SQL says "yesterday" computed from the current time produces a different answer every time you run it, so it can never be safely re-run or backfilled. Research §5.2 names this the number one DAG anti-pattern: never use datetime.now() in pipeline logic. Take the date range from the run, not from the machine.
The task instance lifecycle
Inside a run, each task gets one task instance row, and that row carries a state. The states are the vocabulary you answer failure questions with.
| state | what happened | did it consume compute | what you do about it |
|---|---|---|---|
| queued | the scheduler accepted it, no worker has picked it up | no | nothing, unless it sits there (starved pool) |
| running | a worker is executing it right now | yes | watch the duration against its SLA |
| success | it finished and reported no error | yes | check the rows it wrote before you believe it |
| failed | it raised, and it has no retries left | yes | this is the incident |
| upstream_failed | it never ran because something it depends on failed | no | fix the upstream, not this task |
| skipped | a branch or condition told the scheduler not to run it | no | confirm the branch logic was right |
One failure at the top of a graph writes one failed row and a pile of upstream_failed rows. Counting raw failures without splitting those two states is how a one-task outage gets reported as a ten-task outage.
Common mistake: reading started_at as the answer to "what data is this?". It is the answer to "when did the machine work". A catchup or backfill run makes the two disagree by weeks, and a query that groups by started_at instead of the interval silently attributes February's data to March.
Interview nuance: "what happens when a task fails mid-DAG?" is answered with states, not with vendor names. The failing task goes to failed, everything downstream of it goes to upstream_failed, independent branches keep running, and the run itself is marked failed. Say that and you have answered the question for Airflow, Dagster, Prefect, and Step Functions at once.
On a real platform this differs. Airflow 3.0 (GA April 2025) removed
execution_dateentirely: the concepts are now logical date plus data interval, and backfills are scheduler-managed and first class. Prep material written before 2025 still saysexecution_date, so saying it dates you. The tables here are named generically because the reasoning transfers: Dagster tracks the same idea as asset partitions, and Step Functions carries it as execution input.
CREATE TABLE orchestrator_runs (
run_id TEXT,
dag_id TEXT,
logical_date TEXT, -- the interval this run is FOR; equals data_interval_start on a scheduled run
data_interval_start TEXT, -- inclusive start of the time range the run owns
data_interval_end TEXT, -- exclusive end; a daily run for 2026-03-13 owns [03-13, 03-14)
run_type TEXT, -- scheduled | backfill | manual
state TEXT, -- success | failed (a real dag_run also holds queued and running)
started_at TEXT, -- wall-clock start; far from data_interval_end on a backfill
ended_at TEXT
);
INSERT INTO orchestrator_runs (run_id, dag_id, logical_date, data_interval_start, data_interval_end, run_type, state, started_at, ended_at) VALUES
('ds_2026_03_11', 'daily_sales', '2026-03-11 00:00:00', '2026-03-11 00:00:00', '2026-03-12 00:00:00', 'scheduled', 'success', '2026-03-13 01:40:00', '2026-03-13 01:58:00'),
('ds_2026_03_13', 'daily_sales', '2026-03-13 00:00:00', '2026-03-13 00:00:00', '2026-03-14 00:00:00', 'scheduled', 'success', '2026-03-14 02:05:00', '2026-03-14 02:24:00'),
('ds_2026_03_14', 'daily_sales', '2026-03-14 00:00:00', '2026-03-14 00:00:00', '2026-03-15 00:00:00', 'scheduled', 'failed', '2026-03-15 02:05:00', '2026-03-15 02:19:00'),
('ds_2026_03_15', 'daily_sales', '2026-03-15 00:00:00', '2026-03-15 00:00:00', '2026-03-16 00:00:00', 'scheduled', 'success', '2026-03-16 02:06:00', '2026-03-16 02:27:00'),
('ds_2026_03_09', 'daily_sales', '2026-03-09 00:00:00', '2026-03-09 00:00:00', '2026-03-10 00:00:00', 'manual', 'success', '2026-03-16 11:00:00', '2026-03-16 11:18:00'),
('ds_2026_02_01', 'daily_sales', '2026-02-01 00:00:00', '2026-02-01 00:00:00', '2026-02-02 00:00:00', 'backfill', 'success', '2026-03-16 09:00:00', '2026-03-16 09:21:00'),
('he_2026_03_15_08', 'hourly_events', '2026-03-15 08:00:00', '2026-03-15 08:00:00', '2026-03-15 09:00:00', 'scheduled', 'success', '2026-03-15 09:00:30', '2026-03-15 09:04:30'),
('he_2026_03_15_09', 'hourly_events', '2026-03-15 09:00:00', '2026-03-15 09:00:00', '2026-03-15 10:00:00', 'scheduled', 'success', '2026-03-15 10:00:30', '2026-03-15 10:04:30'),
('he_2026_03_15_10', 'hourly_events', '2026-03-15 10:00:00', '2026-03-15 10:00:00', '2026-03-15 11:00:00', 'scheduled', 'failed', '2026-03-15 11:00:30', '2026-03-15 11:05:20'),
('he_2026_03_15_11', 'hourly_events', '2026-03-15 11:00:00', '2026-03-15 11:00:00', '2026-03-15 12:00:00', 'scheduled', 'failed', '2026-03-15 12:00:30', '2026-03-15 12:05:10'),
('he_2026_01_20_03', 'hourly_events', '2026-01-20 03:00:00', '2026-01-20 03:00:00', '2026-01-20 04:00:00', 'backfill', 'success', '2026-03-16 09:30:00', '2026-03-16 09:34:00'),
('wd_2026_03_02', 'weekly_digest', '2026-03-02 00:00:00', '2026-03-02 00:00:00', '2026-03-09 00:00:00', 'scheduled', 'success', '2026-03-09 03:00:00', '2026-03-09 03:17:00'),
('wd_2026_03_09', 'weekly_digest', '2026-03-09 00:00:00', '2026-03-09 00:00:00', '2026-03-16 00:00:00', 'scheduled', 'failed', '2026-03-16 03:00:00', '2026-03-16 03:18:00');
CREATE TABLE task_instances (
run_id TEXT,
task_id TEXT,
state TEXT, -- success | failed | upstream_failed | skipped (queued and running are transient)
try_number INTEGER, -- 1 on the first attempt; higher means a retry ran
duration_sec INTEGER -- 0 on a task that never actually executed
);
INSERT INTO task_instances (run_id, task_id, state, try_number, duration_sec) VALUES
('ds_2026_03_11', 'extract_orders', 'success', 1, 231),
('ds_2026_03_11', 'clean_orders', 'success', 1, 494),
('ds_2026_03_11', 'load_facts', 'success', 1, 358),
('ds_2026_03_11', 'publish_metrics', 'success', 1, 88),
('ds_2026_03_13', 'extract_orders', 'success', 1, 212),
('ds_2026_03_13', 'clean_orders', 'success', 1, 486),
('ds_2026_03_13', 'load_facts', 'success', 1, 351),
('ds_2026_03_13', 'publish_metrics', 'success', 1, 94),
('ds_2026_03_14', 'extract_orders', 'success', 1, 205),
('ds_2026_03_14', 'clean_orders', 'failed', 2, 618),
('ds_2026_03_14', 'load_facts', 'upstream_failed', 1, 0),
('ds_2026_03_14', 'publish_metrics', 'upstream_failed', 1, 0),
('ds_2026_03_15', 'extract_orders', 'success', 1, 221),
('ds_2026_03_15', 'clean_orders', 'success', 2, 502),
('ds_2026_03_15', 'load_facts', 'success', 1, 366),
('ds_2026_03_15', 'publish_metrics', 'skipped', 1, 0),
('ds_2026_03_09', 'extract_orders', 'success', 1, 198),
('ds_2026_03_09', 'clean_orders', 'success', 1, 470),
('ds_2026_03_09', 'load_facts', 'success', 1, 340),
('ds_2026_03_09', 'publish_metrics', 'success', 1, 91),
('ds_2026_02_01', 'extract_orders', 'success', 1, 260),
('ds_2026_02_01', 'clean_orders', 'success', 1, 505),
('ds_2026_02_01', 'load_facts', 'success', 1, 372),
('ds_2026_02_01', 'publish_metrics', 'success', 1, 99),
('he_2026_03_15_08', 'ingest_events', 'success', 1, 141),
('he_2026_03_15_08', 'rollup_events', 'success', 1, 96),
('he_2026_03_15_09', 'ingest_events', 'success', 1, 138),
('he_2026_03_15_09', 'rollup_events', 'success', 1, 101),
('he_2026_03_15_10', 'ingest_events', 'failed', 2, 275),
('he_2026_03_15_10', 'rollup_events', 'upstream_failed', 1, 0),
('he_2026_03_15_11', 'ingest_events', 'failed', 2, 269),
('he_2026_03_15_11', 'rollup_events', 'upstream_failed', 1, 0),
('he_2026_01_20_03', 'ingest_events', 'success', 1, 132),
('he_2026_01_20_03', 'rollup_events', 'success', 1, 89),
('wd_2026_03_02', 'build_digest', 'success', 1, 930),
('wd_2026_03_02', 'send_digest', 'success', 1, 47),
('wd_2026_03_09', 'build_digest', 'success', 1, 962),
('wd_2026_03_09', 'send_digest', 'failed', 2, 58);-- The first thing you look at on call: the state of every task in the run that broke.
SELECT task_id,
state,
try_number,
duration_sec
FROM task_instances
WHERE run_id = 'ds_2026_03_14'
ORDER BY task_id;Apply
Your turn
The task this lesson builds to.
Write a query that returns each DAG's run health next to the damage inside its runs, as (dag_id, runs, failed_runs, success_pct, real_failures, blocked_tasks), least healthy first, over orchestrator_runs(run_id, dag_id, logical_date, data_interval_start, data_interval_end, run_type, state, started_at, ended_at) joined to task_instances(run_id, task_id, state, try_number, duration_sec) on run_id.
runs is that DAG's number of runs and failed_runs its number of runs in state 'failed', so neither may be inflated by the join. success_pct is the percentage of its runs that succeeded, rounded to 2 decimals. real_failures counts task instances in state 'failed' and blocked_tasks counts task instances in state 'upstream_failed', which are the incidents and the symptoms respectively. Alias the columns exactly, ordered by success_pct ascending and then dag_id.
3 hints and 1 automated check are waiting in the workspace.
Practice
Make it stick
A second problem on the same idea, plus 3 bonus drills.
Write a query that returns every run that processed a data interval more than 1 day older than the day it started, as (run_id, dag_id, data_interval_end, started_at, lag_days), biggest lag first, over orchestrator_runs.
lag_days is the whole number of days from the calendar date of data_interval_end to the calendar date of started_at. Keep only rows where lag_days is greater than 1, so a scheduled run that merely started a day late does not count as a catchup. Alias the columns exactly, ordered by lag_days descending and then run_id.
3 hints and 1 automated check are waiting in the workspace.