dheerajturaga commented on code in PR #73746:
URL: https://github.com/apache/airflow/pull/73746#discussion_r4210751260
##########
task-sdk/src/airflow/sdk/definitions/taskgroup.py:
##########
@@ -702,6 +710,80 @@ def _sort_via_pass_numbering(
sorted_indices = sorted(range(n), key=lambda i: (pass_of[i], i))
return [nodes[i] for i in sorted_indices]
+ def _find_dependency_cycles(self, *, group_dict: dict[str, TaskGroup]) ->
list[list[str]]:
+ """
+ Find children that depend on each other in a cycle when each child
TaskGroup is one unit.
+
+ An edge into any task of a child group counts as an edge into the
group, so a path that
+ leaves a group and comes back into it is a cycle even though the
task-level graph is
+ acyclic. This is stricter than the ordering ``topological_sort``
needs, which only looks
+ at edges into a group's roots.
+
+ :return: one list of child node ids per cycle, each in insertion order
+ """
+ nodes = list(self.children.values())
+ id_to_idx = {nid: i for i, nid in enumerate(self.children)}
+ projected = [
+ self._project_upstream_ids(i, self._get_unit_upstream_ids(child),
id_to_idx, group_dict)
+ for i, child in enumerate(nodes)
+ ]
+ members: dict[int, list[str]] = {}
+ for i, component in
enumerate(self._find_projection_components(projected)):
+ members.setdefault(component, []).append(nodes[i].node_id)
+ return [node_ids for node_ids in members.values() if len(node_ids) > 1]
+
+ @staticmethod
+ def _get_unit_upstream_ids(child: DAGNode) -> Collection[str]:
+ if not isinstance(child, TaskGroup):
+ return child._topological_upstream_ids
+ upstream_ids = set(child._topological_upstream_ids)
+ upstream_ids.update(edge_id for task in child for edge_id in
task.upstream_task_ids)
Review Comment:
Good catch, reproduced: `tg.expand(param=[1, 2, 3])` with a child on
`trigger_rule="always"` raised the `ValueError` from
`MappedTaskGroup.__iter__`. Fixed: the group's unit edges now come from its
`upstream_task_ids`, its non-None `upstream_group_ids` and the
`upstream_task_ids` of each task in `iter_tasks()`, as you suggested. Added
`test_acyclic_task_groups_do_not_warn` in the SDK tests, which turns the
warning into an error and covers that shape, the bridge moved into the group,
and a setup and teardown pair with the work inside the group.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @dheerajturaga before posting
##########
airflow-core/src/airflow/dag_processing/dagbag.py:
##########
@@ -491,7 +496,23 @@ def bag_dag(self, dag: DAG):
:raises: AirflowDagCycleException if a cycle is detected.
:raises: AirflowDagDuplicatedIdException if this dag already exists in
the bag.
"""
- dag.check_cycle()
+ from airflow.sdk.exceptions import TaskGroupCycleDeprecationWarning #
noqa: SDK001
+
+ self.task_group_cycle_warnings.pop(dag.dag_id, None)
+ with warnings.catch_warnings(record=True) as captured_warnings:
+ # DeprecationWarning is ignored by default outside __main__, which
would hide it here too.
+ warnings.simplefilter("always", TaskGroupCycleDeprecationWarning)
+ dag.check_cycle()
+ for captured in captured_warnings:
+ if issubclass(captured.category, TaskGroupCycleDeprecationWarning):
+ self.task_group_cycle_warnings[dag.dag_id] =
str(captured.message)
Review Comment:
Reproduced both orders with two files sharing a Dag id. Fixed: `bag_dag` now
keeps the message in a local and sets or pops `task_group_cycle_warnings` only
after `_add_to_bag` succeeds, so a rejected duplicate (or a Dag a cluster
policy skips) never touches it.
`test_dag_warnings_task_group_cycle_ignores_rejected_duplicate` covers both
orders.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @dheerajturaga before posting
--
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]
