Hi Airflow community, I would like to start a discussion about allowing a single task to require slots from more than one Airflow pool. This follows issue #13975:
https://github.com/apache/airflow/issues/13975 The issue already has maintainer feedback that this change should go through the AIP process because it affects core scheduling and must remain correct with multiple schedulers. I agree. This email is intended to start that discussion before any production implementation PR is opened. Problem A task currently has one pool and one pool_slots value. That works when the task uses one constrained resource, but it cannot describe a task that needs several independently limited resources at the same time. For example, a transfer task may need one connection from a source database pool and two connections from a destination database pool. Other tasks may use either database independently. A synthetic pool for the source-and-destination combination would not share capacity correctly with those tasks, and creating a pool for every possible resource combination does not scale. Proposed user interface I propose adding a mapping-valued pool_requirements argument to operators and TaskFlow decorators: Existing declarations would remain valid and retain their current behavior: Internally, the existing form would be equivalent to one requirement: A mapping avoids the length and ordering problems that would come with separate pools and pool_slots lists. Scheduling semantics A task instance should move to queued only when every required pool: - Exists - Is compatible with the task’s DAG team - Has the requested capacity The scheduler should check the complete requirement set inside its existing pool-locking critical section. It should never reserve one pool while waiting for another. Once the task passes all pool, DAG, task, executor, and team checks, the scheduler can queue it and debit every required pool in the same scheduling decision. A failed check or rolled-back transaction must leave no partial allocation. Slot occupancy should be calculated separately for each pool. This matters for include_deferred: a deferred task may continue to occupy one required pool while releasing another if the two pools have different settings. Storage and compatibility The current scalar pool and pool_slots columns are used by task-instance creation, pool statistics, dependency checks, scheduler filtering, APIs, history, and the UI. Replacing them in one release would require a large backfill and could make mixed-version operation unsafe. My proposed compatibility model is to: - Add normalized task-instance pool-requirement rows - Retain the scalar columns as a compatibility projection - Write normalized rows for new or refreshed task instances - Read the existing scalar values as one requirement when normalized rows are absent - Reject or feature-gate multiple-pool DAGs until every active scheduler understands the new representation This avoids a synchronous full-table backfill of task_instance. Pool accounting can combine normalized rows with legacy rows that do not yet have a normalized representation, without double counting them. Scope The change would cover: - Task SDK - DAG serialization - Mapped operators - Task-instance creation and retry history - Pool accounting - PoolSlotsAvailableDep - Scheduler candidate filtering - Public API responses and filters - UI task details - Metrics - Upgrade documentation The first version would not support: - Dynamic pool acquisition during task execution - Automatic requirements derived from Connections - Alternative requirements such as “A or B” - Preemption - A new fairness policy Multi-resource tasks can make the scheduler’s existing starvation behavior more visible. AIP-100 is already discussing scheduler queueing starvation and priorities: https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/406618462/AIP-100%2BEliminate%2BScheduler%2BQueueing%2BStarvation%2BOn%2BConcurrency%2BLimits I think this proposal can proceed independently if it defines one reusable pool-eligibility path that later scheduler work can call. I do not propose changing priority or fairness behavior as part of multiple-pool support. Questions for the community 1. Is pool_requirements: Mapping[str, int] the right public API, or would a mapping-valued pools argument be clearer? 2. Does the normalized storage plus legacy fallback approach fit Airflow’s migration and compatibility expectations? 3. Should an existing task-instance pool filter match any required pool, or should multiple-pool filtering use a new filter during the compatibility period? 4. Should Airflow impose a small hard limit on the number of pools one task can require? I would benchmark limits before proposing a final number. 5. Is a temporary feature gate sufficient protection against older schedulers enforcing only the scalar compatibility pool? 6. Should the implementation proceed alongside AIP-100, or wait for its scheduler-priority decisions? Related work Two open PRs touch adjacent pool behavior but do not implement this feature: - #69770 changes post-deferral pool occupancy: https://github.com/apache/airflow/pull/69770 - #71220 proposes a cluster-wide include_deferred override: https://github.com/apache/airflow/pull/71220 If the direction makes sense, I will turn the result of this discussion into a formal Confluence AIP. I will also prepare a proof of concept for the occupancy query and scheduler filtering, with PostgreSQL, MySQL, SQLite, and concurrent-scheduler tests, before proposing production implementation PRs. AI disclosure: I used OpenAI Codex to research and draft this proposal. Thanks, Ramachandra Nalam GitHub: @vamsi-klu
