You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hkc-8010 (via GitHub)" <gi...@apache.org> on 2026/07/15 16:54:12 UTC
[PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
hkc-8010 opened a new pull request, #69933:
URL: https://github.com/apache/airflow/pull/69933
When a TaskGroup depends on another TaskGroup through a list (`[group_a, group_b] >> group_c`)
or through a task-level dependency that crosses into another group's entry task, the grid/graph
view sorts the groups as if there were no dependency at all. `_project_child_deps` only looks at
a group's own `upstream_task_ids`, which stays empty in both cases: group-to-group edges only
populate `upstream_group_ids`, and a task-level edge into another group's entry task never
touches the group object itself.
This also pulls in a group's `upstream_group_ids` and its root tasks' upstream ids before
projecting sibling dependencies, in both the serialization-layer sort (`airflow-core`) and the
design-time sort (`task-sdk`) - they share the same algorithm since the #67288/#67688
performance rewrite.
closes: #65291
related: #67964 and #65639 both attempted this fix earlier but were written against the
pre-rewrite `topological_sort` implementation (Kahn's-algorithm style), which no longer exists
after #67288 and #67688 replaced it with the sweep/pass-numbering projection used today. Both
were closed for inactivity/never reviewed rather than merged.
## Changes
- `airflow-core/src/airflow/serialization/definitions/taskgroup.py`: `_project_child_deps` also
considers group-to-group edges (`upstream_group_ids`) and cross-group task-level edges (root
tasks' `upstream_task_ids`)
- `task-sdk/src/airflow/sdk/definitions/taskgroup.py`: same fix, mirrored design-time sort
- Tests for both new scenarios, plus two existing tests whose hardcoded expected orderings were
corrected (they were encoding the old, buggy order)
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Claude Code)
Generated-by: Claude Code following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69933:
URL: https://github.com/apache/airflow/pull/69933#discussion_r3664547998
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py:
##########
@@ -200,15 +200,22 @@ def get_dag_structure(
run_ids = list(session.scalars(dag_runs_select_filter))
task_group_sort = get_task_group_children_getter()
+ latest_group_dict = latest_dag.task_group.get_task_group_dict()
Review Comment:
> I added a call-level memo dict to achieve the same end result as the previous LRU cache (that is problematic for other reasons).
Would it be better to add explicit comment like:
```python
# Intentionally uncached: the latest_group_dict is a derived view of a mutable tree (a cache would go stale with no invalidation) and would pin the whole map for the group's lifetime, fighting the grid endpoint's streaming/expunge.
# Callers build it once per render and pass it down instead.
```
At least I wonder why not add the `lru_cache` at method level then I realize we will introduce the above regressions if we add the cache.
##########
airflow-core/tests/unit/utils/test_task_group.py:
##########
@@ -1187,6 +1220,79 @@ def test_topological_sort_serialized_layered():
)
+def test_topological_group_dep_list_syntax():
+ """List-based deps (`[b0, b1] >> a`) must produce the same topological order as individual deps.
+
+ Declaring a group dependency via a list (`groups >> a`) only populates
+ `upstream_group_ids`, not `upstream_task_ids`, so `a` must not sort as if it had no
+ upstream at all.
+ """
+ with DAG("test_dag_list_dep", schedule=None, start_date=DEFAULT_DATE) as dag:
+ with TaskGroup("a") as tg_a:
+ EmptyOperator(task_id="task")
+
+ groups = []
+ for x in range(3):
+ with TaskGroup(f"b_{x}") as tg_b:
+ EmptyOperator(task_id="task")
+ groups.append(tg_b)
+
+ groups >> tg_a # list-based dep — previously produced the wrong order
+
+ order = [node.node_id for node in dag.task_group.topological_sort()]
+ a_idx = order.index("a")
+ assert all(order.index(f"b_{x}") < a_idx for x in range(3)), (
+ f"Expected all b_x before a in topological order, got: {order!r}"
+ )
+
+
+def test_topological_sort_serialized_list_dep_between_groups():
+ """Same as test_topological_group_dep_list_syntax, exercised on the serialized variant."""
+ with DAG("test_dag_list_dep_serialized", schedule=None, start_date=DEFAULT_DATE) as dag:
+ with TaskGroup("a"):
+ EmptyOperator(task_id="task")
+
+ groups = []
+ for x in range(3):
+ with TaskGroup(f"b_{x}") as tg_b:
+ EmptyOperator(task_id="task")
+ groups.append(tg_b)
+
+ groups >> dag.task_group.children["a"]
+
+ serialized = create_scheduler_dag(dag)
+ order = [node.node_id for node in serialized.task_group.topological_sort()]
+ a_idx = order.index("a")
+ assert all(order.index(f"b_{x}") < a_idx for x in range(3)), (
+ f"Expected all b_x before a in topological order, got: {order!r}"
+ )
+
+
+def test_topological_sort_serialized_task_level_cross_group_dep():
+ """Task-level deps between groups are respected for ordering after serialization.
+
+ A task-level dependency that crosses into another group's entry task (bypassing any
+ group-to-group edge) must still order the downstream group after the upstream one.
+ """
+ with DAG("test_cross_group_task_dep", schedule=None, start_date=DEFAULT_DATE) as dag:
+ with TaskGroup("stage_b"):
+ b_start = EmptyOperator(task_id="b_start")
+ b_end = EmptyOperator(task_id="b_end")
+ b_start >> b_end
+
+ with TaskGroup("stage_a"):
+ a_start = EmptyOperator(task_id="a_start")
+ a_end = EmptyOperator(task_id="a_end")
+ a_start >> a_end
+
+ b_end >> a_start
+
+ serialized = create_scheduler_dag(dag)
+ order = [node.node_id for node in serialized.task_group.topological_sort()]
Review Comment:
Non-blocking nit: We can construct as `dict[task_id, index]` instead of having a list to avoid `O(n)` `.index` call. (Though this only save several ms.)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #69933:
URL: https://github.com/apache/airflow/pull/69933#discussion_r3624557643
##########
airflow-core/tests/unit/utils/test_task_group.py:
##########
@@ -719,25 +721,27 @@ def section_2(value2):
assert dag.task_dict["section_1.section_2.task_4"].downstream_task_ids == {"task_end"}
# Node IDs test
+ # task_start feeds section_1.task_1 directly (a task-level dep crossing into the
+ # group), so section_1 must sort after task_start, not before it.
node_ids = {
"id": None,
"children": [
+ {"id": "task_start"},
{
"id": "section_1",
"children": [
+ {"id": "section_1.task_1"},
+ {"id": "section_1.task_2"},
{
"id": "section_1.section_2",
"children": [
{"id": "section_1.section_2.task_3"},
{"id": "section_1.section_2.task_4"},
],
},
- {"id": "section_1.task_1"},
Review Comment:
cc @uranusjr Since I distinctly remember TP fixing this
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #69933:
URL: https://github.com/apache/airflow/pull/69933
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #69933:
URL: https://github.com/apache/airflow/pull/69933#issuecomment-5101206325
I added a call-level memo dict to achieve the same end result as the previous LRU cache (that is problematic for other reasons).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69933:
URL: https://github.com/apache/airflow/pull/69933#discussion_r3649699560
##########
airflow-core/src/airflow/serialization/definitions/taskgroup.py:
##########
@@ -248,9 +255,24 @@ def topological_sort(self) -> list[DAGNode]:
return self._sweep_projection(nodes, projected)
def _project_child_deps(
- self, child_idx: int, child: DAGNode, id_to_idx: dict[str, int]
+ self,
+ child_idx: int,
+ child: DAGNode,
+ id_to_idx: dict[str, int],
+ group_dict: dict[str | None, SerializedTaskGroup],
) -> tuple[int, ...]:
- upstream_ids = child.upstream_task_ids
+ if isinstance(child, SerializedTaskGroup):
Review Comment:
Why is this needed? This class _is_ SerializedTaskGroup, and this should always return True? When does the `else` block kick in?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on code in PR #69933:
URL: https://github.com/apache/airflow/pull/69933#discussion_r3605630793
##########
airflow-core/src/airflow/serialization/definitions/taskgroup.py:
##########
Review Comment:
Good catch, confirmed this was O(G^2) — task_group_to_dict() recurses into every nested group and calls topological_sort() on each independently, and the Task SDK's own DAG.topological_sort() does the same.
Couldn't pass the map through as a literal parameter since topological_sort() is a public zero-arg method invoked externally by several independent recursive callers (task_group_to_dict, dag_edges, DAG.topological_sort()) — changing its signature would ripple into all of them for no real gain. Instead memoized get_task_group_dict() itself with methodtools.lru_cache, keyed per instance: dag.task_group is the same object for every nested group in one DAG, so the first call builds the map and every subsequent nested call is an O(1) hit. Applied the same fix in task-sdk.
One wrinkle: methodtools.lru_cache has no type stubs, so decorating get_task_group_dict directly widened its return type to Any for every caller, which actually broke mypy downstream in dag_edges.py. Kept the memoized logic behind a private helper with an explicitly-typed public wrapper so mypy still trusts the declared return type.
Added a test asserting the cache is actually hit (not rebuilt) across nested topological_sort() calls, in both packages.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69933:
URL: https://github.com/apache/airflow/pull/69933#discussion_r3661684012
##########
airflow-core/tests/unit/utils/test_task_group.py:
##########
@@ -719,25 +721,27 @@ def section_2(value2):
assert dag.task_dict["section_1.section_2.task_4"].downstream_task_ids == {"task_end"}
# Node IDs test
+ # task_start feeds section_1.task_1 directly (a task-level dep crossing into the
+ # group), so section_1 must sort after task_start, not before it.
node_ids = {
"id": None,
"children": [
+ {"id": "task_start"},
{
"id": "section_1",
"children": [
+ {"id": "section_1.task_1"},
+ {"id": "section_1.task_2"},
{
"id": "section_1.section_2",
"children": [
{"id": "section_1.section_2.task_3"},
{"id": "section_1.section_2.task_4"},
],
},
- {"id": "section_1.task_1"},
Review Comment:
As pointed out above, a rewrite has changed things a bit (including test cases). The fix proposed here is logically correct.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #69933:
URL: https://github.com/apache/airflow/pull/69933#issuecomment-5102789110
### Backport failed to create: v3-3-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/30349625361'> Run details </a>
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>v3-3-test</td>
<td><a href="https://github.com/apache/airflow/commit/d7aa92913bfed7b37081a3962fb6072053fc8d75"><img src='https://img.shields.io/badge/Commit-d7aa929-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker d7aa929 v3-3-test
```
This should apply the commit to the v3-3-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
If you don't have cherry-picker installed, see the [installation guide](https://github.com/apache/airflow/blob/main/dev/README.md#how-to-backport-pr-with-cherry-picker-cli).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #69933:
URL: https://github.com/apache/airflow/pull/69933#issuecomment-5103383724
Manually backport in https://github.com/apache/airflow/pull/70591.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "hkc-8010 (via GitHub)" <gi...@apache.org>.
hkc-8010 commented on code in PR #69933:
URL: https://github.com/apache/airflow/pull/69933#discussion_r3605629028
##########
airflow-core/tests/unit/utils/test_task_group.py:
##########
@@ -493,17 +493,11 @@ def test_task_group_to_dict_and_dag_edges(dag_maker):
nodes = task_group_to_dict(dag.task_group)
edges = dag_edges(dag)
+ # group_d depends on group_c (`group_d << group_c`), so it must sort after group_c,
+ # not before task1 as it did prior to the #65291/#67964 topological-sort fix.
Review Comment:
Done, reworded to describe the scenario being tested instead of citing the issue number.
##########
airflow-core/tests/unit/utils/test_task_group.py:
##########
@@ -1178,6 +1183,80 @@ def test_topological_sort_serialized_layered():
)
+def test_topological_group_dep_list_syntax():
+ """List-based deps (`[b0, b1] >> a`) must produce the same topological order as individual deps.
+
+ Regression test for apache/airflow#65291: declaring a group dependency via a list
Review Comment:
Done here too. Fixed the same pattern in two other spots in this file that had the same issue but weren't directly flagged.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix grid/graph view topological sort for group-level and cross-group dependencies [airflow]
Posted by "viiccwen (via GitHub)" <gi...@apache.org>.
viiccwen commented on code in PR #69933:
URL: https://github.com/apache/airflow/pull/69933#discussion_r3595126396
##########
airflow-core/tests/unit/utils/test_task_group.py:
##########
@@ -493,17 +493,11 @@ def test_task_group_to_dict_and_dag_edges(dag_maker):
nodes = task_group_to_dict(dag.task_group)
edges = dag_edges(dag)
+ # group_d depends on group_c (`group_d << group_c`), so it must sort after group_c,
+ # not before task1 as it did prior to the #65291/#67964 topological-sort fix.
Review Comment:
Could we mention the intention of the testing instead of using issue number?
##########
airflow-core/tests/unit/utils/test_task_group.py:
##########
@@ -1178,6 +1183,80 @@ def test_topological_sort_serialized_layered():
)
+def test_topological_group_dep_list_syntax():
+ """List-based deps (`[b0, b1] >> a`) must produce the same topological order as individual deps.
+
+ Regression test for apache/airflow#65291: declaring a group dependency via a list
Review Comment:
same here.
##########
airflow-core/src/airflow/serialization/definitions/taskgroup.py:
##########
Review Comment:
This rebuilds the complete Dag-level TaskGroup mapping every time `topological_sort()` is called. The Grid / Graph serializers recursively call `topological_sort()` for every nested TaskGroup, so a dag with G non-empty groups now performs approximately G full traversals of the same G-group tree. That changes the overall group-sorting cost to $O(G^2)$.
Could we build the mapping once for the entire recursive render and pass it through to each nested `topological_sort()` call? The same in the Task SDK's `topological_sort()` path.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org