bingqin2 commented on issue #75865:
URL: https://github.com/apache/airflow/issues/75865#issuecomment-6608042457
I tested three mapped-task fixes on 3.4.0rc2 against 3.3.2, using
`dag.test()` on SQLite with Python 3.12, in clean virtualenvs installed from
PyPI with the matching constraints. All three bugs reproduce on 3.3.2 and are
fixed in the RC.
- #73960: the Dag from #55225. An unmapped short-circuit returns False,
followed by `dummy2 >> dummy3.expand(x=[1, 2, 3]) >> publish`, all with
`none_failed`. On 3.3.2, `dummy3[0]`, `[1]` and `[2]` succeed. On the RC all
three are skipped. `dummy2` and `publish` are skipped on both.
- #74283, which also covers #73959: the Dag from #74175, `gate >> a >>
[b_all_done, b_none_failed]` in a task group expanded over `[True, False]`. On
3.3.2 nothing at map index 1 is skipped. On the RC, `a[1]`, `b_all_done[1]` and
`b_none_failed[1]` are skipped, and map index 0 still runs every task.
- #69075: the Dag from #69036. On 3.3.2, a task group expanded over `["14"]`
hands its consumer the bare string, which iterates as `['1', '4']`. On the RC
the consumer gets `['14']`. With two expansions, both versions give `['15',
'23']`.
<details>
<summary>Dags and scripts</summary>
`dags/rc_69075_mapped_group_single_expansion.py`
```python
# #69075 (issue #69036): the output of a mapped task group consumed outside
the group must be a
# list with one element per expansion, also when the group expands only once.
import json
import os
import pendulum
from airflow.sdk import dag, task, task_group
@dag(schedule=None, start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
catchup=False)
def rc_69075_mapped_group_single_expansion():
@task
def convert(a):
return a
@task
def do_something(a):
return a
@task_group
def group(a):
return {"test": do_something(a)}
@task
def consume(dtm_result, label):
with open(os.environ["RC_OUT"], "a") as f:
f.write(json.dumps({"input": label, "received":
list(dtm_result)}) + "\n")
for label, values in (("single", ["14"]), ("two", ["15", "23"])):
result = group.override(group_id=f"group_{label}").expand(
a=convert.override(task_id=f"convert_{label}")(values)
)
consume.override(task_id=f"consume_{label}")(result["test"], label)
the_dag = rc_69075_mapped_group_single_expansion()
```
`dags/rc_73960_short_circuit_expanded_mapped.py`
```python
# #73960 (issue #55225): an unmapped short-circuit must skip a downstream
task that is mapped over a
# literal list (its instances already exist), even with
trigger_rule="none_failed".
import pendulum
from airflow.sdk import dag, task
@dag(
schedule=None,
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
catchup=False,
default_args={"trigger_rule": "none_failed"},
)
def rc_73960_short_circuit_expanded_mapped():
@task.short_circuit(ignore_downstream_trigger_rules=True)
def dummy1():
return False
@task
def dummy2():
return "dummy2"
@task
def dummy3(x: int):
return f"dummy3-{x}"
@task
def publish():
return "published"
dummy1() >> dummy2() >> dummy3.expand(x=[1, 2, 3]) >> publish()
the_dag = rc_73960_short_circuit_expanded_mapped()
```
`dags/rc_74283_mapped_group_short_circuit.py`
```python
# #74283 (issue #74175): a short-circuit inside a mapped task group must
skip every later task of
# that map index, including ones whose trigger rule accepts a skipped
upstream.
import pendulum
from airflow.sdk import dag, task, task_group
@dag(schedule=None, start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
catchup=False)
def rc_74283_mapped_group_short_circuit():
@task_group
def group(value):
@task.short_circuit # ignore_downstream_trigger_rules=True (default)
def gate(v):
return v
@task
def a():
pass
@task(trigger_rule="all_done")
def b_all_done():
pass
@task(trigger_rule="none_failed")
def b_none_failed():
pass
first = a()
gate(value) >> first
first >> [b_all_done(), b_none_failed()]
group.expand(value=[True, False])
the_dag = rc_74283_mapped_group_short_circuit()
```
`check.py`
```python
import json
import os
import runpy
import sys
from pathlib import Path
import airflow
from airflow.models import TaskInstance
from airflow.utils.session import create_session
print(f"== apache-airflow {airflow.__version__}")
out = Path(os.environ["RC_OUT"])
for path in
sorted(Path(os.environ["AIRFLOW__CORE__DAGS_FOLDER"]).glob("rc_*.py")):
dag = next(v for v in runpy.run_path(str(path)).values() if
type(v).__name__ == "DAG")
out.write_text("")
dr = dag.test()
with create_session() as session:
tis = session.query(TaskInstance).filter_by(dag_id=dag.dag_id,
run_id=dr.run_id).all()
states = sorted((ti.task_id, ti.map_index, str(ti.state)) for ti in
tis)
print(f"-- {dag.dag_id}: run {dr.state}")
for task_id, map_index, state in states:
print(f" {task_id}[{map_index}] {state}")
for line in out.read_text().splitlines():
print(" consumer got", json.loads(line))
```
`setup.sh`
```bash
#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
for v in 3.3.2 3.4.0rc2; do
[ -x "venv_$v/bin/airflow" ] && { echo "venv_$v exists"; continue; }
uv venv -q -p 3.12 "venv_$v"
VIRTUAL_ENV="$PWD/venv_$v" uv pip install -q "apache-airflow==$v" \
--constraint
"https://raw.githubusercontent.com/apache/airflow/constraints-$v/constraints-3.12.txt";
--prerelease=allow
echo "venv_$v: $(venv_$v/bin/python -c 'import airflow, airflow.sdk;
print(airflow.__version__, airflow.sdk.__version__)')"
done
```
`run.sh`
```bash
#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
for v in 3.3.2 3.4.0rc2; do
export AIRFLOW_HOME=$PWD/home_$v AIRFLOW__CORE__DAGS_FOLDER=$PWD/dags
AIRFLOW__CORE__LOAD_EXAMPLES=False RC_OUT=$PWD/out_$v.jsonl
rm -rf "$AIRFLOW_HOME"; mkdir -p "$AIRFLOW_HOME"
"venv_$v/bin/airflow" db migrate > "migrate_$v.log" 2>&1
"venv_$v/bin/python" check.py 2> "check_$v.log" | grep -E "^(==|--| )"
done
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
