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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
