[WIP] AIP-117 Dynamic resolution of task fields

Status

Motivation

A task’s pool, pool_slots and priority_weight are fixed in the Dag source. There is no supported way to decide, when triggering a run, that specific tasks of that run should use a different pool, a different number of slots, or a different priority. There is also no way for a Dag author to make those values depend on a run parameter, so a Dag that needs two different pool assignments has to be written twice.
Nothing proposed here is impossible today. A task_instance_mutation_hook cluster policy can already rewrite these values per run, and https://github.com/apache/airflow/pull/68198 (3.3) added dag_run to that hook so deployments can route on dag_run.conf (policies.py:71-91). What this AIP adds is authorization, per-run scope, an audit record, and a 400 at trigger time in place of a task instance that never schedules.
These attributes are deliberately not templated. https://github.com/apache/airflow/pull/29821 made it an error to list any BaseOperator.__init__ parameter except email in template_fields, because the scheduler needs these values before the task starts, while Jinja and XCom values only exist after it starts. Any solution therefore has to supply the value as data, at or before run creation.
The request recurs and has been answered consistently. https://github.com/apache/airflow/discussions/47860 drew a maintainer reply that this is impossible today because pools are selection criteria inside the scheduler’s query, that it is a serious architectural change, and that it needs an AIP and probably a proof of concept. https://github.com/apache/airflow/issues/35542 drew the design sketch: not Jinja, because the scheduler must not execute code, but a declarative mapping resolved without running user code. Related requests: https://github.com/apache/airflow/issues/35803, https://github.com/apache/airflow/issues/33657, https://github.com/apache/airflow/issues/35689. No existing AIP covers this.
Following the guidance in that discussion, a proof of concept is a precondition for the vote rather than a follow-up. Implementation detail beyond what is needed to judge the design belongs in that pull request rather than on this page.

Considerations

The contentious question is not the mechanism but whether users should choose a pool at all. https://github.com/apache/airflow/discussions/43235 answered no for backfill, on the grounds that pools are deployment-level resource protection. The answer here is an administrator-declared envelope rather than a blanket user right.

What change do you propose to make?

Two phases, both in scope for this AIP:
Trigger-time overrides. A new optional field on the run-creation request, with its validation, authorization, storage and application to task instances, plus an editor in the trigger form.
Binding to Dag params. A way for a Dag author to declare that one of these attributes takes its value from a Dag param, so that scheduled runs benefit too, plus UI to show which attributes are bound and what they resolved to. Phase 2 also replaces the coarse phase 1 permission with a purpose-built one.
Everything else is future work, listed at the end of this section.

What problem does it solve?

Attributes
pool, pool_slots and priority_weight. These are the three values the scheduler uses to admit and rank work, and all three are already copied onto the task instance row when it is created (taskinstance.py:780-801), so no new scheduler input is introduced.
queue and executor are excluded because they select the worker fleet rather than the amount of work allowed: [sdk] queue_to_coordinator maps a queue name to a coordinator spec that determines the runtime and, under KubernetesExecutor, the worker pod’s service account, secrets and image, and the security model names queue separation as the Celery isolation measure (security_model.rst:284-288). Allowing either would let a user holding only “can trigger this Dag” move existing code onto a fleet with different credentials. run_as_user (impersonation) and executor_config (an arbitrary pod specification) are excluded permanently for the same reason. max_tries is excluded because it accumulates across clears and needs its own retry-accounting argument.
Phase 1: the request field
One optional field on TriggerDAGRunPostBody (datamodels/dag_run.py:220-233). That model subclasses StrictBaseModel with extra="forbid", so the field has to be declared explicitly and threaded through validate_context.
{"task_overrides": {
  "extract": {"pool": "heavy_pool"},
  "transform": {"pool": "gpu_pool", "pool_slots": 4}
}}
Keys are exact task_ids or group_ids. A group_id applies to every task in that group, including nested groups, and an entry for a specific task_id takes precedence over an entry for a group containing it. Patterns such as extract_* are not supported, because the scheduler cannot re-check a pattern’s match set when it later applies the override; group keys cover the many-tasks case instead.
Storage. A nullable JSON column on dag_run, next to conf rather than inside it, since conf is deep-merged into the Dag params and validated against them. Storing the payload on the run is what makes an override survive a clear or a retry, because clear_task_instances re-runs refresh_from_task (taskinstance.py:408), which otherwise resets the task instance to the task’s own values.
Validation happens in SerializedDAG.create_dagrun (serialization/definitions/dag.py:571), the single choke point for run creation, next to the existing params validation at :679-680. It short-circuits when the field is absent, so unmodified triggers issue no extra query. The checks are: the config flag is on (otherwise 4xx, not a silent ignore); every key names a task or group that exists in the Dag; every named pool exists, is an approved override target, and belongs to a compatible team, resolved in one WHERE pool IN (...) query rather than one per key; pool_slots is a positive integer no larger than the target pool’s slots, above which the task instance could never be scheduled; and priority_weight is an integer. Failures return 400, using one message for “does not exist”, “not an approved target” and “belongs to another team”, with the specific reason logged server-side. Distinguishing them in the response would make the endpoint a pool and team enumeration oracle for a caller who holds nothing but trigger permission.
Application happens each time a task instance is created or refreshed, which includes a clear, a verify_integrity pass for a newly observed Dag version, and mapped expansion. One path needs care: the non-noop branch of _get_task_creator (dagrun.py:2010-2018) constructs a TaskInstance whose __init__ calls refresh_from_task without a dag_run, so an override applied there would be dropped silently. That branch is selected whenever any task_instance_mutation_hook is installed, for any purpose, so it is a common configuration rather than an edge case. Every application point must re-validate the override and, on failure, fall back to the task’s own value and log, rather than write a value that reaches queued and can never be scheduled. Re-authorization is deliberately not attempted, because no principal exists in the scheduling loop; the config flag and the pool opt-in are re-checked instead, so revoking either one stops overrides already stored on live runs.
Surfaces. Of the ten create_dagrun call sites, phase 1 covers two: the REST and UI trigger route (routes/public/dag_run.py:795), which also reaches airflow-ctl through its generated datamodel, and asset materialisation (routes/public/assets.py:501), which needs extra work because that route lists create_dagrun keyword arguments explicitly and authorizes inline rather than through a dependency. The rest are deferred: backfill (rejected in discussion 43235), DAG.test() and airflow tasks test (local debugging, no principal), scheduler-created runs (no request body, which is what phase 2 addresses), and the Execution API path used by TriggerDagRunOperator. That last one is deferred on authorization grounds: the caller is a task JWT with no per-Dag or pool authorization, those routes have no action_logging, and triggering_user_name is inherited from the parent run, so an override chosen by task code would be attributed to whoever triggered the parent.
Phase 2: binding to Dag params
A marker object in the Task SDK, written here as Lookup although the final name is open, lets the Dag author name a param instead of a literal:
HeavyOperator(task_id="crunch", pool=Lookup.param("target_pool"))
At run creation the marker resolves against the run’s merged and validated params, which is a dictionary lookup rather than evaluation of user code. Resolution reuses the phase 1 machinery: the same application points, the same re-validation, and the same fallback to the statically declared value when the named pool is unavailable. Adding a marker can therefore never make a working Dag fail, which matters because the same code path runs for scheduled runs where nobody is present to read an error.
Phase 2 is what makes scheduled and asset-triggered runs configurable, since they have no request body. It also lets the author constrain the choice, because a param declared with Param(enum=[...]) gives an author-controlled allowlist that the existing param schema machinery validates at trigger time.
The serialized form is an ordered list of sources even though phase 2 implements only one kind, so that a fallback chain such as “this param, otherwise this Variable, otherwise the declared value” can be added later without a format change. Beyond that, the marker has to survive Dag serialization, for which VariableInterval (definitions/deadline.py:386-421) is precedent as an SDK-side object already resolved server-side inside create_dagrun; the Dag parser has to reject a marker naming a param that does not exist or has an incompatible type, so authors find out at parse time; and the accepted argument types for these three parameters have to widen to allow the marker, in BASEOPERATOR_ARGS_EXPECTED_TYPES and in the task decorator’s copy of the same check.
Phase 2 also carries the auth-manager work described in Security, and adds an indication on the task instance detail page of which attributes were bound to a param and what they resolved to, so a value differing from the Dag source is explainable without reading the run payload.
Future work
Further source kinds for the marker, such as Airflow Variables, subject to the constraints in Security. Persistent per-Dag override rules, which would cover scheduled runs without editing the Dag at all. Overrides for queue and executor, if a worker-fleet authorization story is developed. Backfill, unless discussion 43235 is revisited. XCom can never be a source for these attributes, because it resolves after the point at which the scheduler needs the value.

Why is it needed?

Are there any downsides to this change?

Discussion 43235 decided not to allow pool overrides in Airflow 3 backfill, because pools are deployment-level resource protection. This AIP does not reopen backfill, and it answers that reasoning by having the administrator declare the envelope rather than granting users a blanket right.
In phase 1, sending an override requires pool-administration rights, which is more privilege than the operation needs. That is the cost of avoiding an auth-manager interface change in the first phase, and phase 2 removes it.
An override is invisible in the Dag source, which is a real debugging surprise. It is mitigated by showing the values on the run and task instance detail pages, and by logging whenever an override changes a value.
Precedence, highest first: the cluster policy, then a trigger-time override, then a param binding, then the value written in the Dag, then the Airflow default. The cluster policy only wins today because the hook happens to run last (taskinstance.py:987). This AIP must make that ordering a stated requirement with a test on every creation path, and must surface the case where a policy overrode a requested value. Otherwise, in a deployment whose policy sets pool unconditionally, the new field would silently do nothing.
Guardrails. The administrator declares the envelope through a boolean column on slot_pool, default false, so only nominated pools are legal targets and default_pool is not one of them. pool_slots is capped by the target pool’s own slots. priority_weight has no numeric cap, because the value is only meaningful relative to other task instances; the control is that its effect is confined to ordering within a pool the user was already authorized to use. [core] allow_dag_run_task_overrides, mirroring the existing dag_run_conf_overrides_params, defaults to False and is enforced both at run creation and at every application point. The cluster policy still runs last.

Which users are affected by the change?

Dag authors need change nothing for phase 1, which is the point, since making an attribute dynamic today would mean editing every task that needs it. Phase 2 is opt-in for authors who want it. Deployment Managers get a DB migration, a config flag to enable, and pools to nominate. API, CLI and UI users get one new optional request field. Implementers of third-party auth managers are unaffected in phase 1 and gain one optional method in phase 2, whose default preserves existing behaviour.

How are users affected by the change? (e.g. DB upgrade required?)

A DB upgrade is required: one nullable JSON column on dag_run and one boolean column on slot_pool, in a single migration with a downgrade() that drops both. Both columns are nullable with no server default, so components running pre-migration code during an upgrade window read NULL as “no overrides” rather than erroring, and the JSONB variant is declared up front, which makes the change metadata-only on PostgreSQL and MySQL 8 or later. Downgrading discards stored payloads, while task instances already materialised keep the values they were given. Phase 2 adds no column and no serialization version bump, because a Dag that uses no marker serializes exactly as it does today.

What is the level of migration effort (manual and automated) needed for the users to adapt to the breaking changes? (especially in context of Airflow 3)

None. There are no breaking changes, and with the config flag off nothing changes on upgrade: run airflow db migrate and change no Dags. Phase 2’s auth-manager addition is designed so that an auth manager which does not implement it behaves exactly as it did in phase 1.

Other considerations?

Security
Validation is not authorization. create_dagrun is the right place to validate, but it has no principal. Authorization needs a body-aware dependency on the trigger route, an inline check next to is_authorized_dag in materialize_asset (assets.py:459-471), and 4xx rejection rather than silent acceptance on any surface with no authenticated end-user principal. Declaring the pydantic field authorizes nothing by itself.
Authorization in two steps. requires_access_dag(method="POST", access_entity=DagAccessEntity.RUN) resolves dag_id from path and query parameters only, so a pool name supplied in the body is invisible to it, and a body-aware dependency is needed either way. Requiring two checks on one endpoint is established practice (routes/ui/dags.py:92-97 requires four).
Phase 1 uses the existing is_authorized_pool with a write method, so a caller must hold pool-administration rights in addition to Dag trigger rights. This needs no interface change and no work in the FAB provider, which keeps the first phase small, at the cost of demanding more privilege than the operation warrants and offering no per-Dag or per-team granularity.
Phase 2 introduces a purpose-built method, is_authorized_task_override(*, details: DagDetails, pool: str, pool_slots: int, user), called once per distinct pool in the payload. Its default implementation in BaseAuthManager delegates to the phase 1 check rather than returning False, which is the important detail: no deployment’s behaviour changes on upgrade, third-party and older FAB auth managers keep working untouched, and a manager that wants per-Dag granularity overrides the method. A deny-by-default body would instead disable the feature for every deployment that had not yet implemented it. The delegate target is stricter than the eventual permission, so the permissive-looking default is still fail-closed in practice. Two alternatives are rejected: an @abstractmethod breaks third-party subclasses at instantiation, and a new DagAccessEntity member is source-compatible but functionally breaking, because FAB raises on an unmapped entity with no handler upstream, so new core against an older FAB provider returns 500 instead of 403.
Pools are a resource guardrail. The escalation the permission exists to prevent is moving a task out of a small administrator-set pool into a large existing one, which “only pools that already exist” does nothing to stop. Hence the pool opt-in and the slot cap. priority_weight sits in the same envelope, because ranking ahead of other tenants in a shared pool is the same class of decision. The audit record already exists: both target routes carry Depends(action_logging()), which writes a Log row before the handler runs, capturing the request body with per-key redaction, including for rejected requests.
Phase 2 is lower precedence but not lower privilege. Run conf is deep-merged into the run’s params before validation, so a task declaring pool=Lookup.param("tier") could have its pool chosen by any caller who can trigger the Dag, by passing conf={"tier": ...}, with no permission and no config flag, and on the surfaces phase 1 deliberately excludes. Phase 2 must close this, either by requiring the named param to be constrained by Param(enum=[...]) and refusing conf-supplied values outside it, or by putting a resolved value through the same permission, pool opt-in, slot cap and team checks as a request-supplied one. This is the main security question the phase 2 design has to settle.
Values, not references. The request payload carries literal values only. If it could name an Airflow Variable, anyone able to trigger a Dag would gain a read-any-Variable primitive, by pointing an override at a secret and reading the value back off the task instance row or the UI, and nothing on the run-creation path masks secrets, because masking lives in the worker-side fetch helpers. The same reasoning bounds future marker sources: a Variable source is acceptable only if the resolved value is masked or never persisted, and an environment-variable source is rejected outright, because resolution would run in the API server and scheduler, which hold the JWT signing key and database credentials that a Dag author is not entitled to read.
Multi-team. The scheduler enforces pool ownership at scheduler_job_runner.py:881-893, but only when the pool has a team, so global pools are usable by every team and moving a team Dag’s task into one is an explicit decision (the recommendation is to refuse it). Trigger-time validation has to replicate that check and fail closed, unlike the scheduler’s team lookup, which returns {} on error. It cannot defer to PoolSlotsAvailableDep, which repeats the check but is IGNORABLE = True, nor to the scheduler, because Pool.slots_stats aggregates by pool name with no team dimension, so a mis-scoped override reaching queued consumes another team’s slots.
Architecture boundaries
No user code runs anywhere new. The phase 1 payload is inert data, and phase 2 resolution is a dictionary lookup into params the API server has already merged and validated. models/referencemixin.py:36-38 states the rule this respects: the scheduler-side mirror of the SDK ResolveMixin has resolve() removed because the scheduler should not resolve references.
No component boundary moves. There is no new worker database access, no new Execution API surface, and no Dag File Processor involvement beyond phase 2’s parse-time check on marker validity. The resolved value lands in the task instance row the worker already receives, so the Task SDK runtime is untouched.
Performance
Overrides resolve once per run into a task_id to values mapping, memoised on the DagRun, and are then looked up per task instance. That is O(tasks) per materialisation and independent of map_index cardinality. Matching per task instance instead would be O(entries times task instances), which for 50 entries against a 10,000-index expansion is 500,000 matches. On the paths that run inside the scheduling loop, resolution must add no queries.
The bulk insert fast path is preserved. _create_task_instances uses bulk_insert_mappings when the cluster policy is a noop and bulk_save_objects otherwise (dagrun.py:2079). Applying overrides inside insert_mapping, which already receives dag_run, keeps the faster path intact.
The critical section is unchanged, with one thing left to measure. No new join, predicate or index is introduced: TaskInstance.pool stays a plain indexed column (ti_pool), and Pool.slots_stats never reads dag_run. Pool existence is checked once at trigger time, not per scheduling loop. However, TI.dag_run is lazy="joined" at mapper level (taskinstance.py:690), so every select(TI) materialises the full run row, and the proof of concept has to measure whether the new column matters there before choosing between deferring the column and using a side table. The payload is bounded regardless: at most 50 entries, 8 kB serialized, and 5 distinct pool names.
User interface
The phase 1 editor goes in the Advanced Options accordion of the trigger form (ConfigForm.tsx:98-124), with pool values from the existing usePoolServiceGetPools hook as a dropdown, following RunBackfillForm.tsx as the precedent for structured fields that are not Dag params. The stored payload is shown next to conf on the run detail page, and the effective values next to the pool rows on the task instance detail page. Note that MappedTaskInstance/Details.tsx:92-102 reads pool from the serialized task, so it would otherwise display a static value next to an overridden one. Changing the request model also means regenerating the OpenAPI spec, the UI and airflow-ctl clients, and adding en i18n keys.

What defines this AIP as "done"?

Phase 1:
The optional field on the trigger and asset-materialise routes, authorized by Dag trigger rights plus is_authorized_pool, with 4xx wherever no authenticated end-user principal exists, and a non-disclosing 400 for an unknown key, an unapproved or cross-team pool, pool_slots above the pool total, or a non-integer priority_weight.
The slot_pool opt-in column and the dag_run payload column in one migration, with a working downgrade().
Application at every task instance creation path, including the non-noop cluster-policy branch, with the bulk insert demonstrably still in use; re-validation with fallback, covering an override invalidated or the flag disabled after run creation; and the cluster policy demonstrably running last on every path.
[core] allow_dag_run_task_overrides defaulting to off, the payload bounds, the trigger-form editor and the detail-page displays with en i18n keys.
Docs for pools and Dag runs, and a cross-reference from the cluster-policies page explaining when to use which mechanism.
Tests: the API routes; model tests with a non-noop task_instance_mutation_hook installed so the per-object branch is exercised; a scheduler test proving that the overridden pool is what Pool.slots_stats accounts against; cross-team rejection and the team-to-global-pool case; migration up and down.
Performance gates: identical query counts with and without a payload around verify_integrity and _do_scheduling, and unchanged critical-section timing for a run whose override targets a saturated pool.
Phase 2:
The marker type, accepted for all three attributes, serialized as an ordered source list, surviving a Dag serialization round trip, with parse-time rejection of an unknown or wrongly typed param.
Resolution at the same application points, with fallback to the declared value and a Dag warning rather than an exception when a pool is unavailable, so a scheduled run is never blocked.
is_authorized_task_override with a default that delegates to the phase 1 check, a SimpleAuthManager override, a compatibility test against a manager that does not implement it, and a significant-change note for the interface addition.
The conf-supplied param question from Security settled and implemented, so a param binding cannot bypass the phase 1 checks.
The task instance detail page showing which attributes were bound and how they resolved, plus docs and tests covering a scheduled run and a mapped task.