Fix grid/graph view topological sort for group-level and cross-group dependencies - #69933
Conversation
uranusjr
left a comment
There was a problem hiding this comment.
I added a couple of commits to hoist common logic into the shared lib. This looks good enough to me now.
Also, I intentionally removed methodtools.lru_cache from get_task_group_dict since an SDK task group can change shape between calls and cause the cache to become stale. The serialized object is generally short-lived (the canonical source is in the database), so the cache is not generally too useful anyway.
4521039 to
2266042
Compare
…deps TaskGroup._project_child_deps only looked at a group's own upstream_task_ids, which stays empty for a direct group-to-group dependency (list or individual `>>`) and for a task-level dependency that crosses into another group's entry task. Both cases sorted the group as if it had no upstream at all. Now also pulls in the group's upstream_group_ids and its root tasks' upstream task ids before projecting sibling dependencies. Applied to both the serialization-layer sort and the mirrored design-time sort in task-sdk. closes: apache#65291 Related: apache#67964 (closed for inactivity, written against the topological_sort implementation before PR apache#67288/apache#67688 rewrote it) and apache#65639 (draft, same issue, also predates the rewrite).
…ent not issue numbers viiccwen pointed out that fetching the group map inside topological_sort() rebuilds the whole DAG's group tree on every nested group's own call, turning a render with G groups into an O(G^2) cost. get_task_group_dict() is now memoized per DAG instance (kept behind a small private helper since methodtools.lru_cache has no type stubs and would otherwise widen the public method's return type to Any for every caller). Also reworded test comments/docstrings that cited issue numbers to describe what's actually being verified instead.
The cache these tests asserted was removed in the previous commit, so the identity check and the _get_task_group_dict_cached.cache_info() assertions no longer apply.
2b20511 to
dedf545
Compare
|
I added a call-level memo dict to achieve the same end result as the previous LRU cache (that is problematic for other reasons). |
There was a problem hiding this comment.
#70573 was just merged -- which should resolve all the provider test failures in last run.
jason810496
left a comment
There was a problem hiding this comment.
The patch LGTM, my comments are non-blocking and they won't effect the prod code.
Backport failed to create: v3-3-test. View the failure log Run detailsNote: As of Merging PRs targeted for Airflow 3.X In matter of doubt please ask in #release-management Slack channel.
You can attempt to backport this manually by running: cherry_picker d7aa929 v3-3-testThis should apply the commit to the v3-3-test branch and leave the commit in conflict state marking After you have resolved the conflicts, you can continue the backport process by running: cherry_picker --continueIf you don't have cherry-picker installed, see the installation guide. |
|
Manually backport in #70591. |
…dependencies (#69933) (#70591) * Fix grid/graph view topological sort for group-level and cross-group deps TaskGroup._project_child_deps only looked at a group's own upstream_task_ids, which stays empty for a direct group-to-group dependency (list or individual `>>`) and for a task-level dependency that crosses into another group's entry task. Both cases sorted the group as if it had no upstream at all. Now also pulls in the group's upstream_group_ids and its root tasks' upstream task ids before projecting sibling dependencies. Applied to both the serialization-layer sort and the mirrored design-time sort in task-sdk. closes: #65291 Related: #67964 (closed for inactivity, written against the topological_sort implementation before PR #67288/#67688 rewrote it) and #65639 (draft, same issue, also predates the rewrite). * Address review feedback: cache get_task_group_dict, describe test intent not issue numbers viiccwen pointed out that fetching the group map inside topological_sort() rebuilds the whole DAG's group tree on every nested group's own call, turning a render with G groups into an O(G^2) cost. get_task_group_dict() is now memoized per DAG instance (kept behind a small private helper since methodtools.lru_cache has no type stubs and would otherwise widen the public method's return type to Any for every caller). Also reworded test comments/docstrings that cited issue numbers to describe what's actually being verified instead. * Hoist common logic into shared lib * Remove caching on get_task_group_dict * Remove stale get_task_group_dict cache tests The cache these tests asserted was removed in the previous commit, so the identity check and the _get_task_group_dict_cached.cache_info() assertions no longer apply. * Add call-level task group memo to reduce calc * Tidy Typy * Add test for task group memoing --------- Co-authored-by: TP <uranusjr@apache.org> Co-authored-by: LIU ZHE YOU <zhu424.dev@gmail.com> (cherry picked from commit d7aa929) # Conflicts: # airflow-core/src/airflow/api_fastapi/core_api/services/ui/task_group.py Co-authored-by: Hemkumar Chheda <95332229+hkc-8010@users.noreply.github.com>
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_depsonly looks ata group's own
upstream_task_ids, which stays empty in both cases: group-to-group edges onlypopulate
upstream_group_ids, and a task-level edge into another group's entry task nevertouches the group object itself.
This also pulls in a group's
upstream_group_idsand its root tasks' upstream ids beforeprojecting sibling dependencies, in both the serialization-layer sort (
airflow-core) and thedesign-time sort (
task-sdk) - they share the same algorithm since the #67288/#67688performance rewrite.
closes: #65291
related: #67964 and #65639 both attempted this fix earlier but were written against the
pre-rewrite
topological_sortimplementation (Kahn's-algorithm style), which no longer existsafter #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_depsalsoconsiders group-to-group edges (
upstream_group_ids) and cross-group task-level edges (roottasks'
upstream_task_ids)task-sdk/src/airflow/sdk/definitions/taskgroup.py: same fix, mirrored design-time sortcorrected (they were encoding the old, buggy order)
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code following the guidelines
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.