[hussein-awala]

When the triggerer picks up newly deferred tasks, _create_workload resolves the run's Dag version and fetches the full serialized Dag row separately for every trigger, then deserializes the whole Dag just to check one task's start_from_trigger flag (triggerer_job_runner.py#L950-L960). A batch is usually many tasks of a few runs, so the same work repeats dozens of times on the supervisor's main loop, which also persists trigger events and heartbeats the job. Instead, read each Dag's latest version and load each distinct Dag version once per pickup batch.

related: #76209

The rule that picks a run's version is unchanged and keeps the run-pinned semantics from #69988: the run's pinned version (the latest for runs without one), falling back to the task instance's own dag_version_id when that version has no serialized Dag. The rule moves into DBDagBag._resolve_version_for_run, which takes the latest-version lookup as a callable, so DBDagBag and the batch lookup share it.

The batch lookup is a new BatchSerializedDagResolver in airflow/models/dagbag.py, next to DBDagBag and marked :meta private: like it. Keeping it in that module means _resolve_version_for_run stays private and nothing outside dagbag.py calls it; the triggerer only creates a resolver and calls its resolve(). The long-lived DBDagBag caches are untouched; there is no new configuration or schema change.

One visible difference: among the task triggers claimed in one loop (at most [triggerer] max_trigger_to_select_per_loop, 50 by default), unversioned runs of a Dag all resolve against the latest version read at the first lookup. A version rewritten in place by SerializedDagModel.write_dag mid-batch is not re-read for that batch's later triggers; the next batch reads fresh values. The batch's triggers share one deserialized Dag, used only for get_task().

_create_workload gains an optional keyword-only serdag_resolver argument that build_trigger_workloads passes, so overrides must accept it. It no longer calls DBDagBag._version_from_dag_run, so a DBDagBag subclass overriding it is bypassed here; nothing in the tree overrides it. DBDagBag.get_serialized_dag_model_for_run, which #69988 added for this call site, has no caller left and is removed. A follow-up planned in #76209 builds on this PR to keep deserialized Dags across batches; this PR stands on its own.

In a local benchmark draining one Dag run's deferred tasks through load_triggers():

Scenario	Statements	Before (ms)	After (ms)
50 triggers, one batch, SQLite	110 → 12	84.3	18.5
50 triggers, one batch, PostgreSQL 16	110 → 12	238.0	49.9
200 triggers, four batches, SQLite	440 → 48	868.5	45.7
200 triggers, four batches, PostgreSQL 16	440 → 48	1734.9	163.0

Ten of the 12 statements per batch are claim and bulk-fetch queries every batch already runs. This is the best case: the saving shrinks as a batch spans more distinct Dags or versions.

Statement counts are engine-wide, exclude COMMIT, and were identical in every round; the listener only increments a counter, so timings carry no per-statement instrumentation cost. Times are medians of 5 rounds, each running the base and the change once, alternating which runs first; no before/after ranges overlap. An unchanged control bench (50 failed triggers) moved +0.3 % on SQLite and +24.8 % on PostgreSQL, with overlapping ranges on both. PostgreSQL wall times are noisy: the control's runs spanned 154 to 256 ms. SQLite (file database) and PostgreSQL 16 (Docker, 2-CPU, 8 GiB VM) ran on a macOS Apple-silicon development machine; these are local measurements, not end-to-end latency.

Unit tests in tests/unit/models/test_dagbag.py cover the resolver's memoization, its fallback to the task instance's version, and one latest-version lookup per Dag. A database-backed build_trigger_workloads test in tests/unit/jobs/test_triggerer_job.py expects one version lookup and one serialized Dag fetch for three triggers of one run, and fails without the change.

Validation:

• Focused tests (test_triggerer_job.py, test_trigger.py, test_dagbag.py): 247 passed, 1 skipped on SQLite (the skip is a PostgreSQL-only test).
• Same tests through Breeze with the CI image: 248 passed on PostgreSQL 14; 247 passed, 1 skipped on MySQL 8.0.
• Regular and manual static checks passed on the full diff.
• What breeze verify lists passed on a local merge of the nine PRs opened together under Improve triggerer performance (AI-assisted deep dive) #76209: core DB tests of the API, Always, CLI, Core, Other and Serialization groups on SQLite (12,254 run, no failures), core non-DB tests (5,582 passed) and the apache-airflow documentation build.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 5.5)

Generated-by: Claude Code (Opus 5.5) following the guidelines

🤖 Generated with Claude Code