fix: corrects issues with edge rendering on the graph view

[codecae]

closes: #35301

The graph UI has issues rendering edges between dependent tasks nested in task groups. Edges render with excessive bends in various directions instead of directly between tasks and their parents task groups as they are collapsed and expanded. They also tend to render vertically off center at node borders. This PR addresses this edge rendering issue.

Example code that reproduces the issue:

from typing import Dict, List
import pendulum
from airflow.sdk import BaseOperator, dag, task, task_group
@dag(
    schedule=None,
    start_date=pendulum.datetime(2023, 1, 1),
)
def test_dag():
    _all_tasks: Dict[str, List[BaseOperator]] = {}
    def _generate_task_group_callable(group_id, tasks):
        @task_group(group_id=group_id)
        def _task_group():
            [t() for t in tasks]
        return _task_group
    def _generate_task_callable(task_id, task_set):
        def _generate():
            @task(task_id=task_id)
            def _task():
                pass
            if _all_tasks.get(task_set) is None:
                _all_tasks[task_set] = []
            _all_tasks[task_set].append(_task())
        return _generate
    _set_callables = []
    for _set in range(0, 2):
        _set_id = f"set_{_set}"
        _group_callables_a = []
        for _nga in range(0, 1):
            _nga_id = f"group_{_nga}"
            _group_callables_b = []
            for _ngb in range(0, 3):
                _ngb_id = f"group_{_ngb}"
                _task_callables = []
                for _task in range(0, 10):
                    _task_id = f"task_{_task}"
                    _callable = _generate_task_callable(_task_id, _set_id)
                    _task_callables.append(_callable)
                _group_callable_inner = _generate_task_group_callable(_ngb_id, _task_callables)
                _group_callables_b.append(_group_callable_inner)
            _group_callable_outer = _generate_task_group_callable(_nga_id, _group_callables_b)
            _group_callables_a.append(_group_callable_outer)
        _set_callable = _generate_task_group_callable(_set_id, _group_callables_a)
        _set_callables.append(_set_callable)
    [_s() for _s in _set_callables]
    for i in range(0, len(_all_tasks["set_0"])):
        _all_tasks["set_0"][i] >> _all_tasks["set_1"][i]
test_dag()

Rendering issue examples:

With this fix:

---

[pierrejeambrun]

Thanks for the PR, indeed that need fixing.

[pierrejeambrun]

Instead of doing another full pass on filteredEdges and creating another big list of edges with many duplicated, move this deduplication step directly in the formatChildNode function. This will allow to not insert duplicated edges at all in the first place, removing the need for an extra step in the process.

[codecae]

Instead of doing another full pass on filteredEdges and creating another big list of edges with many duplicated, move this deduplication step directly in the formatChildNode function. This will allow to not insert duplicated edges at all in the first place, removing the need for an extra step in the process.

The deduplication logic has been moved to formatChildNode with reduced complexity.

[pierrejeambrun]

Thanks, looking good!

[github-actions]

Backport successfully created: v3-0-test

Status	Branch	Result
✅	v3-0-test