Tasks with all_done trigger rule execute when they should skip
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow version
Other Airflow 2 version (please specify below)
### If "Other Airflow 2 version" selected, which one?
2.10.5
### What happened?
I have a task that filters a list based on values of a second list. As such it depends on two upstream tasks, each producing one of the lists. The second upstream task is mapped dynamically.
With the default trigger rule `all_success` the task in question will skip if any of the mapped tasks skips, which feels counter-intuitive.
With the `all_done` trigger rule the task behaves as expected, waiting for all mapped tasks to finish and executing with the resulting mapped list of outputs.
However, if _all_ of the mapped tasks skip, which is a valid result in this context, the tasks still tries to execute.
This might be related to #51320.
### What you think should happen instead?
The task receives the original list and instead of the second list with filter values it receives `None`. I think this happens because the mapped return values of skipped tasks (`None`) are reduced to `None`.
In my opinion it would be more appropriate to reduce them to `[]`.
Apart from that, the task should just be skipped.
### How to reproduce
```python
from random import random
from typing import Any, List
import pendulum
from airflow.decorators import dag, task
from airflow.exceptions import AirflowSkipException
from airflow.models.param import Param
@dag(
"playground",
"Try things",
schedule=None,
start_date=pendulum.now(),
params={
"p_abort": Param(
type="number",
title="p_abort",
minimum=0.0,
maximum=1.0,
description="Probability to abort mapped tasks",
)
},
)
def playground():
@task.python
def produce_list() -> List[int]:
return list(range(25))
@task.python
def maybe_abort(value: int, **kwargs) -> int:
if random() > kwargs["params"]["p_abort"]:
return value
raise AirflowSkipException()
@task.python
def filter_list(values1: List[int], values2: List[int]) -> List[int]:
# if values1 is None or values2 is None:
# raise AirflowSkipException()
return [v for v in values1 if v in values2]
values1 = produce_list()
values2 = maybe_abort.expand(value=values1)
filter_list.override(trigger_rule="all_done")(values1, values2)
playground()
if __name__ == "__main__":
playground().test()
```
This issue can be worked around with the commented check, but I don't think we should have to validate task inputs.
### Operating System
Ubuntu 24.04
### Versions of Apache Airflow Providers
_No response_
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start by running the provided playground DAG with every dynamically mapped maybe_abort task raising AirflowSkipException. Trace how dynamic task mapping reduces skipped outputs and how the all_done trigger rule handles the downstream filter_list task. Done means the all-skipped case reduces inputs appropriately and skips the downstream task instead of executing it.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100