ashb commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4317359083
##########
airflow-core/docs/authoring-and-scheduling/loops.rst:
##########
@@ -152,9 +152,23 @@ definition.
- The previous iteration's result, or ``None`` in iteration 0.
* - ``loop.result``
- The terminal body task's result (the XCom pushed under the key
``return_value``) for the current iteration, used by the gate.
+ * - ``loop.group_id``
+ - The group id of the loop.
Review Comment:
Isn't this already available on the ti?
##########
airflow-core/src/airflow/models/task_coordinates.py:
##########
@@ -839,3 +926,18 @@ def select_producer_ids(
region_index=request.region_index,
session=self.session,
)
+
+
+def get_task_instance_execution_position(
+ task_instance: TaskInstance, *, session: Session
+) -> TaskInstanceExecutionPosition:
+ """
+ Place a stored task instance in its mapped expansion and loop pass.
+
+ This is what the Execution API sends a worker when the task starts
(``TIRunContext.loop`` and
+ ``TIRunContext.expanded_ti_count``), so server-side consumers such as
listeners reporting a manual state
+ change describe the task instance as the worker would, without reading
region rows themselves.
+ """
+ return TaskCoordinateResolver(DBDagBag(load_op_links=False),
session).get_execution_position(
Review Comment:
Using a new dag bag here feels expensive - this should be reworked to not
need a dagbag - either pass an existing bag in, or require the dag to be
already loaded
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -657,14 +657,67 @@ def body():
assert response.status_code == 200
if version == "2026-06-30":
assert "loop" not in response.json()
+ assert "expanded_ti_count" not in response.json()
return
assert response.json()["loop"] == {
"node_id": loop.group_id,
"index": 2,
"max_iterations": 4,
"terminal_task_id": loop.terminal_task_id,
"terminal_is_mapped": mapped,
+ "region_id": str(parent.id),
+ "trajectory": 1,
+ "forked_from_region_id": None,
+ "resumes_from_index": 0,
Review Comment:
Maybe these should be absent when not set
##########
airflow-core/tests/unit/models/test_dynamic_region.py:
##########
@@ -1114,3 +1115,17 @@ def
test_loop_sequence_of_mapped_task_skips_passes_a_fork_replaced(loop_run, ses
(True, 3, 0),
(True, 3, 1),
]
+
+
[email protected]("forks", [0, 1, 2])
+def test_trajectory_counts_the_forks_a_region_descends_from(forks):
+ regions = {}
+ region = DynamicRegion(id=uuid4(), dag_id="dag", run_id="run",
node_id="loop")
Review Comment:
Isn't there a make_region helper?
##########
airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py:
##########
@@ -457,6 +457,28 @@ class LoopContext(BaseModel):
terminal_task_id: str
terminal_is_mapped: bool
+ region_id: uuid.UUID
+ """
+ The loop region holding this iteration.
+
+ A mapped task's own ``region_id`` is its expansion's, so this is how its
instances are tied to the
+ unmapped tasks of the same pass.
+ """
+
+ trajectory: int = Field(default=1, ge=1)
+ """
+ 1 for the loop's original region, and one more for each fork of it.
+
+ Clearing a loop with its later iterations forks the loop region, and the
iterations from
+ ``resumes_from_index`` on run again in the fork.
+ """
+
+ forked_from_region_id: uuid.UUID | None = None
+ """The loop region this fork replaces from ``resumes_from_index`` on,
``None`` for an original region."""
+
+ resumes_from_index: int = Field(default=0, ge=0)
+ """The first iteration ``region_id`` holds, 0 for an original region."""
Review Comment:
Do we need to expose this one? Also the doc string is "not useful /correct"
I think
##########
airflow-core/src/airflow/models/task_coordinates.py:
##########
@@ -69,6 +71,40 @@ class _ProducerRequest:
loop_node_id: str | None = None
[email protected](kw_only=True)
+class LoopPosition:
+ """
+ The pass of the innermost loop a task instance runs in.
+
+ ``region_id`` is the loop region that holds the pass, also for a mapped
task, whose own ``region_id`` is
+ its expansion's. A clear that reruns later passes forks the loop region:
``trajectory`` is 1 for the
+ original region and grows by one with each fork, ``forked_from_region_id``
is the region the fork
+ replaced and ``resumes_from_index`` the first pass the fork owns (0 for an
original region).
+ """
+
+ group: SerializedLoopTaskGroup
+ index: int
+ region_id: UUID
+ trajectory: int
+ forked_from_region_id: UUID | None
+ resumes_from_index: int
+
+
[email protected](kw_only=True)
+class TaskInstanceExecutionPosition:
Review Comment:
So we need a new struct for this?
##########
airflow-core/docs/authoring-and-scheduling/loops.rst:
##########
@@ -413,3 +433,18 @@ which leave them in place. To keep later iterations, clear
from the UI or the
REST endpoint.
Tasks you leave uncleared are not recomputed using the results of the rerun.
+
+.. _loops-trajectories:
+
+Reruns and trajectories
+-----------------------
+
+Clearing later iterations starts a new execution of the loop that replaces the
old one from the
+iteration after the rerun gate. Each execution is a trajectory:
``loop.trajectory`` is 1 for the
Review Comment:
I thought trajectory was a uuid in the back end?
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -657,14 +657,67 @@ def body():
assert response.status_code == 200
if version == "2026-06-30":
assert "loop" not in response.json()
+ assert "expanded_ti_count" not in response.json()
return
assert response.json()["loop"] == {
"node_id": loop.group_id,
"index": 2,
"max_iterations": 4,
"terminal_task_id": loop.terminal_task_id,
"terminal_is_mapped": mapped,
+ "region_id": str(parent.id),
+ "trajectory": 1,
+ "forked_from_region_id": None,
+ "resumes_from_index": 0,
Review Comment:
Null rather then 0?
--
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]
