kacpermuda commented on code in PR #74752:
URL: https://github.com/apache/airflow/pull/74752#discussion_r4315004733
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1489,28 +1587,38 @@ def is_dag_run_asset_triggered(
@attrs.frozen
class TaskInstancePosition:
"""
- Where a task instance runs: inside a loop body or not, and the map index
users see.
+ Where a task instance runs: inside a loop body or not, the map index users
see and the loop iteration.
A plain mapped task keeps the region of its own expansion, so its dag,
task, try, logical date and
map index stay unique; inside a loop they repeat on every pass, so only
the execution UUID identifies
- the run.
+ the run. ``loop_index`` is the zero-based iteration and
``loop_max_iterations`` the limit of the loop, the
+ counterparts of ``try_number`` and the maximum number of tries, and
``loop_group_id`` the loop's group id.
+ All three describe the innermost loop and are ``None`` outside a loop. A
stored task whose task isn't
+ attached has no ``loop_max_iterations``, nor ``loop_group_id`` when it is
mapped, and a stored mapped task
+ in a loop has no ``loop_index`` unless ``get_orm_task_instance_positions``
could read its region. A runtime
+ task instance has no ``loop_index`` nor ``loop_max_iterations`` unless
placed with
+ ``get_task_instance_position``, as only its template context holds them.
+ ``TaskInstanceInfo`` explains how these read together with the try and the
region in events.
"""
in_loop: bool
map_index: int
+ loop_index: int | None = None
+ loop_max_iterations: int | None = None
+ loop_group_id: str | None = None
-def _is_task_in_loop(task: Any) -> bool:
+def _get_enclosing_loop(task: Any) -> Any | None:
from airflow.sdk.definitions._internal.loop import LoopTaskGroup
from airflow.sdk.definitions.taskgroup import TaskGroup
from airflow.serialization.definitions.taskgroup import
SerializedLoopTaskGroup, SerializedTaskGroup
group = getattr(task, "task_group", None)
while isinstance(group, (TaskGroup, SerializedTaskGroup)):
if isinstance(group, (LoopTaskGroup, SerializedLoopTaskGroup)):
- return True
+ return group
group = group.parent_group
- return group is not None
+ return None
Review Comment:
Good catch, fixed in 7441d2d: `_find_enclosing_loop` assumes a loop again
when the walk stops on a group of an unknown type, and logs a warning, so the
worst case is an extra runId rather than merged iterations.
--
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]
