GitHub user Ananyaas created a discussion: Proposal : Native Sequential and
Fail-Fast Execution for Dynamic Mapped Tasks
### Summary
I would like to propose a feature for the **Airflow 3.x** roadmap to natively
support **Sequential and Fail-Fast execution** within the Dynamic Task Mapping
framework (`.expand()`).
As Airflow 2.x moves past its April 2026 EOL, defining core task orchestration
boundaries for Airflow 3.x is highly relevant. Currently, Dynamic Task Mapping
operates on a strict parallel MapReduce assumption where all mapped indices
(`map_index=0, 1, 2...`) are independent. However, production workflows
frequently require runtime-generated data chunks to execute chronologically,
where an early failure must immediately halt the remaining sequence.
### The Architectural Problem
While concurrency controls like `max_active_tis_per_dag=1` can force serialized
execution, they fail to create state-based dependencies between individual
sibling indices:
1. **No Cascading Failures:** If `map_index=0` fails, Airflow still triggers
`map_index=1` and `map_index=2`. In data loops where downstream steps assume
the integrity of upstream modifications, this causes resource waste and data
corruption.
2. **Global Fail-Stop Limitations:** The global `fail_stop=True` flag stops the
*entire* DAG. This is too destructive when you only want to break execution for
a specific mapped task array while letting independent branches finish normally.
### Proposed Airflow 3.x API Design
I propose introducing two parameters to the execution mapper: `sequential: bool
= False` and `fail_fast: bool = False`.
```python
@task
def process_data_chunk(chunk_id):
# Core processing logic
pass
# Proposed configuration for Airflow 3.x
process_data_chunk.expand(
chunk_id=runtime_generated_list,
sequential=True,
fail_fast=True
)
```
### Strategic Technical Implementation
Because this modifies core scheduling assumptions, I want to ensure it
integrates seamlessly with Airflow 3.x's scheduler architecture:
1. **Gated Execution (`airflow/models/dagrun.py`):**
Inside `_get_ready_tis()`, if `sequential=True` is enabled, the scheduler
gates the readiness of `map_index = N` until `map_index = N-1` reaches a
terminal `SUCCESS` state.
2. **State Cascading (`airflow/ti_deps/deps/trigger_rule_dep.py`):**
If `fail_fast=True` is active and `map_index = N` switches to `FAILED`, the
scheduling loop automatically cascades all subsequent un-run sibling indices
(`N+1` to max index) directly to `UPSTREAM_FAILED`.
### Seeking Community Feedback
I am prepared to lead this implementation and draft an Airflow Improvement
Proposal (AIP) if required. Before taking this to the devlist, I would love
initial feedback:
* Does this feature align with the scheduling goals of Airflow 3.x?
* Are there edge cases regarding task retries or clearing mapped task states
that we should plan for?
GitHub link: https://github.com/apache/airflow/discussions/70094
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
