uranusjr commented on a change in pull request #21688:
URL: https://github.com/apache/airflow/pull/21688#discussion_r810688105
##########
File path: airflow/utils/dag_cycle_tester.py
##########
@@ -49,33 +42,4 @@ def check_cycle(dag: 'DAG') -> None:
:raises AirflowDagCycleException: If cycle is found in the DAG.
"""
- # default of int is 0 which corresponds to CYCLE_NEW
- visited: Dict[str, int] = defaultdict(int)
- path_stack: Deque[str] = deque()
- task_dict = dag.task_dict
-
- def _check_adjacent_tasks(task_id, current_task):
- """Returns first untraversed child task, else None if all tasks
traversed."""
- for adjacent_task in current_task.get_direct_relative_ids():
- if visited[adjacent_task] == CYCLE_IN_PROGRESS:
- msg = f"Cycle detected in DAG. Faulty task: {task_id}"
- raise AirflowDagCycleException(msg)
- elif visited[adjacent_task] == CYCLE_NEW:
- return adjacent_task
- return None
-
- for dag_task_id in dag.task_dict.keys():
- if visited[dag_task_id] == CYCLE_DONE:
- continue
- path_stack.append(dag_task_id)
- while path_stack:
- current_task_id = path_stack[-1]
- if visited[current_task_id] == CYCLE_NEW:
- visited[current_task_id] = CYCLE_IN_PROGRESS
- task = task_dict[current_task_id]
- child_to_check = _check_adjacent_tasks(current_task_id, task)
- if not child_to_check:
- visited[current_task_id] = CYCLE_DONE
- path_stack.pop()
- else:
- path_stack.append(child_to_check)
+ dag.topological_sort(include_subdag_tasks=False)
Review comment:
Would be nice if we could add a few tests for this to find out.
--
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]
