simpleqt opened a new pull request, #76266:
URL: https://github.com/apache/airflow/pull/76266
## Fix labeled edges to a task in a TaskGroup connecting the whole group
A labeled edge between a task inside a TaskGroup and a task outside of it
connected the whole group instead of only the named tasks:
```python
with DAG("label_grouped_task", schedule=None):
with TaskGroup("transform"):
clean_t = clean()
check()
load_t = load()
clean_t >> Label("rows") >> load_t
```
Before this change, `load` depended on **every leaf** of `transform` (i.e.
on `check` too, via `transform.downstream_join_id`), so a failing `check` put
`load` in `upstream_failed` even though the DAG author only wired `clean` to
`load`. The mirrored spelling `load << Label("rows") << clean` behaved
correctly, and so did `clean_t.set_downstream(load_t, Label("rows"))`, so three
spellings of the same edge disagreed with each other. The mirrored direction
was broken too: `s1 << Label("L") << clean` (with `s1` inside another group)
wired `clean` upstream of the entire `s1` group.
The same behavior has existed since 2.5.2 (introduced by #29410).
### Cause
`EdgeModifier._convert_streams_to_task_groups` promoted plain tasks to their
TaskGroup whenever the two sides of a labeled edge belonged to different
groups. `TaskGroup._set_relatives` fed the group's leaves (not the group) into
the modifier, so:
- a labeled edge from a task in a group was promoted to a group-level edge
(bug fixed here), and
- labeled edges between explicit TaskGroups needed that promotion to keep
their join-id wiring (kept working here).
### Fix
- `TaskGroup._set_relatives` now hands the TaskGroup itself to an
`EdgeModifier` (`task_group >> Label("x") >> other`), so labeled group-to-group
edges keep the exact join-id wiring and edge info they had before (the #17469
behavior is unchanged).
- `EdgeModifier` no longer promotes plain tasks to their TaskGroup, so a
label never changes which tasks a dependency connects: `a >> Label("x") >> b`,
`a << Label("x") << b`, and `a.set_downstream(b, Label("x"))` now all produce
the same single task-level edge, matching the unlabeled edge.
### Tests
- New `TestEdgeModifierTaskInGroup` in
`airflow-core/tests/unit/utils/test_edgemodifier.py`: the three labeled
spellings from a task inside a group, a labeled edge between tasks of two
different groups, and a guard that labeled group-to-group edges still use join
ids. The first three fail on `main` and pass with this change.
- `test_task_group_edge_modifier_chain` in the task-sdk now asserts the
group itself is recorded as upstream, matching the unlabeled `chain(start, tg,
t3)` behavior instead of the old flattened bookkeeping.
Fixes #76170
Signed-off-by: simpleqt <[email protected]>
--
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]
