hussein-awala opened a new pull request, #76215: URL: https://github.com/apache/airflow/pull/76215
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`](https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/jobs/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. - Selected core suites and the documentation build were not run locally; CI validation is pending. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Opus 5.5) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
