apache / apache/airflow

Tasks with all_done trigger rule execute when they should skip

Open
#53,667 10 comments 0 reactions 0 assignees View on GitHub
_eol_affected_version:2.10 area:core area:dynamic-task-mapping kind:bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.