You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kacpermuda (via GitHub)" <gi...@apache.org> on 2026/10/19 18:00:30 UTC
[PR] Expose task instance loop pass and expansion size to listeners [airflow]
kacpermuda opened a new pull request, #75132:
URL: https://github.com/apache/airflow/pull/75132
Listener consumers such as OpenLineage should be able to trace an event back to the task execution that produced it from what the listener gets, not by looking the attempt up by `ti.id` in the metadata database. For mapped tasks and task loops part of that wasn't exposed:
- `TIRunContext.expanded_ti_count`: the number of live instances in the mapped task's expansion, counted within the iteration inside a loop. Without it a listener doesn't know how many events to expect, so it can't tell "3 of 20 instances failed" or that all of them finished.
- `LoopContext.region_id`: the loop region holding the iteration. A mapped task's own `region_id` is its expansion's, so this is what ties it to the other tasks of the same pass.
- `LoopContext.trajectory`, `forked_from_region_id` and `resumes_from_index`: which rerun of the loop an iteration ran in, and which execution it replaced from which iteration on. Before, ordering reruns meant comparing the UUIDv7 region ids of two events.
- Public `RuntimeTaskInstance.loop`, `loop_decision` and `expanded_ti_count`. A gate's decision was only in the private `_loop_decision`, so its listener couldn't tell `until` being met from `max_iterations` being reached.
- `get_task_instance_execution_position(ti, session=...)`, so listeners called on the API server (manual state changes) get the same data as the worker, instead of reading `dynamic_region` rows.
The fields go into the unreleased `2026-10-30` Execution API version, so they can ship with 3.4.0.
related: #74752
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Claude Code)
Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323974118
##########
airflow-core/src/airflow/models/task_coordinates.py:
##########
@@ -243,10 +278,47 @@ def loop_context(
regions.update(
load_region_ancestry([ti.region_id], dag_id=ti.dag_id, run_id=ti.run_id, session=self.session)
)
- position = loop_position(regions, ti.region_id, ti.region_index, group.node_id)
- if position is None:
+ coordinate = DynamicRegion.find_physical_coordinate(
+ ti.region_id, ti.region_index, group.node_id, regions
+ )
+ if coordinate is None:
raise ValueError("Task coordinates do not belong to the pinned loop")
- return group, position[1]
+ region_id, index = coordinate
+ region = regions[region_id]
+ is_fork = region.forked_from_region_id is not None
+ return LoopPosition(
+ group=group,
+ index=index,
+ region_id=region_id,
+ trajectory=get_trajectory(region, regions),
+ forked_from_region_id=region.forked_from_region_id,
+ resumes_from_index=region.resumes_from_index if is_fork else None,
+ )
+
+ def get_expanded_ti_count(
+ self, ti: TaskCoordinate, *, task: SerializedOperator | None = None
+ ) -> int | None:
+ """
+ Count the live instances of a mapped task in the expansion ``ti`` belongs to, None if not mapped.
+
+ This is what a worker gets as ``TIRunContext.expanded_ti_count``.
+ """
+ if ti.region_id == SENTINEL_REGION_ID:
+ return None
+ if task is None:
+ task = self.get_task(ti.dag_id, ti.run_id, ti.task_id, dag_version_id=ti.dag_version_id)
+ if not task.get_needs_expansion():
+ return None
+ return self.session.scalar(
+ select(func.count()).where(
Review Comment:
AIP-104 might mean that this value is incomplete/inaccurate (as it would only say how much has been expanded so far. Note that is unlikely to land in 3.4, so just an FYI.
Yes, as long as it's it's indexed, yes this extra query is fine, and we only run the query for mapped tasks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4326580824
##########
airflow-core/docs/authoring-and-scheduling/loops.rst:
##########
@@ -413,3 +431,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 previous ones from the
+iteration after the rerun gate. Each execution is a trajectory, identified by ``loop.region_id`` and
+numbered by ``loop.trajectory``: 1 for the first one, growing by one with each rerun. The rerun gate, and whatever you clear in its
+iteration, run again in the trajectory that iteration belongs to, with their next try. The
+replacement iterations run in the new trajectory, from ``loop.resumes_from_index``, and their tries
+count from the start again. In the
Review Comment:
```suggestion
count from the 1 again for each task. In the
```
##########
go-sdk/pkg/execution/genmodels/models.gen.go:
##########
@@ -1329,11 +1333,20 @@ type LoopContext struct {
// NodeID corresponds to the JSON schema field "node_id".
NodeID string `msgpack:"node_id"`
+ // RegionID corresponds to the JSON schema field "region_id".
+ RegionID string `msgpack:"region_id"`
Review Comment:
omit empty? `*string`?
##########
go-sdk/pkg/execution/genmodels/models.gen.go:
##########
@@ -1329,11 +1333,20 @@ type LoopContext struct {
// NodeID corresponds to the JSON schema field "node_id".
NodeID string `msgpack:"node_id"`
+ // RegionID corresponds to the JSON schema field "region_id".
+ RegionID string `msgpack:"region_id"`
+
+ // ResumesFromIndex corresponds to the JSON schema field "resumes_from_index".
+ ResumesFromIndex interface{} `msgpack:"resumes_from_index,omitempty"`
Review Comment:
Something is slightly off with our generated schema, as this should be coming out as:
```suggestion
ResumesFromIndex *int `msgpack:"resumes_from_index,omitempty"`
```
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -5270,6 +5307,18 @@
],
"default": null
},
+ "expanded_ti_count": {
+ "anyOf": [
+ {
+ "type": "integer"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Expanded Ti Count"
+ },
Review Comment:
This one should probably be an integer, or absent. i.e. null is not a valid value for this field -- it's sent as an integer, or not sent at all.
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -3266,14 +3266,51 @@
"terminal_is_mapped": {
"title": "Terminal Is Mapped",
"type": "boolean"
+ },
+ "region_id": {
+ "format": "uuid",
+ "title": "Region Id",
+ "type": "string"
+ },
+ "trajectory": {
+ "minimum": 1,
+ "title": "Trajectory",
+ "type": "integer"
+ },
+ "forked_from_region_id": {
+ "anyOf": [
Review Comment:
Yeah, `anyOf` means "1 _or more_ of these schemas" (IMO this is something Pydantic gets fundamentally wrong.) Elsewhere in the FastAPI app we explicitly post-process the schema to fix this. I think we should do the same here.
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -3266,14 +3266,51 @@
"terminal_is_mapped": {
"title": "Terminal Is Mapped",
"type": "boolean"
+ },
+ "region_id": {
+ "format": "uuid",
+ "title": "Region Id",
+ "type": "string"
+ },
+ "trajectory": {
+ "minimum": 1,
+ "title": "Trajectory",
+ "type": "integer"
+ },
+ "forked_from_region_id": {
+ "anyOf": [
Review Comment:
Or could also be a "sent, or absent too".
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -371,19 +371,28 @@ def ti_run(
should_retry=_is_eligible_to_retry(previous_state, ti.try_number, ti.max_tries),
multi_team=conf.getboolean("core", "multi_team"),
)
- resolver = TaskCoordinateResolver(dag_bag, session)
- if loop_context := resolver.loop_context(ti):
- group, index = loop_context
- terminal = resolver.get_task(
- ti.dag_id, ti.run_id, group.terminal_task_id, dag_version_id=ti.dag_version_id
- )
- context.loop = LoopContext(
- node_id=group.node_id,
- index=index,
- max_iterations=group.max_iterations,
- terminal_task_id=group.terminal_task_id,
- terminal_is_mapped=terminal.get_needs_expansion(),
- )
+ if ti.region_id != SENTINEL_REGION_ID:
+ resolver = TaskCoordinateResolver(dag_bag, session)
+ task = resolver.get_task(ti.dag_id, ti.run_id, ti.task_id, dag_version_id=ti.dag_version_id)
+ if (expanded_ti_count := resolver.get_expanded_ti_count(ti, task=task)) is not None:
+ context.expanded_ti_count = expanded_ti_count
+ if (loop := resolver.get_loop_position(ti, task=task)) is not None:
+ terminal = resolver.get_task(
+ ti.dag_id, ti.run_id, loop.group.terminal_task_id, dag_version_id=ti.dag_version_id
+ )
+ context.loop = LoopContext(
+ node_id=loop.group.node_id,
+ index=loop.index,
+ max_iterations=loop.group.max_iterations,
+ terminal_task_id=loop.group.terminal_task_id,
+ terminal_is_mapped=terminal.get_needs_expansion(),
+ region_id=loop.region_id,
+ trajectory=loop.trajectory,
+ )
+ # Set only for a fork, as the route leaves out unset fields
+ if loop.forked_from_region_id is not None:
+ context.loop.forked_from_region_id = loop.forked_from_region_id
+ context.loop.resumes_from_index = loop.resumes_from_index
Review Comment:
Might be worth extracting this all to a `resolver.get_loop_context()`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323833676
##########
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:
You’re right about the docstring: a fork can replace iterations held by several earlier regions, not only the one it was forked from. It now says it’s the previous region in the chain. I’d like to keep the field: it links a rerun to the one before it without relying on any ordering, and it should still hold per loop if nested loops ever come.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4324016733
##########
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:
One follow-up, since that helper existed for a reason. The OpenLineage provider needs the same loop and expansion data for events it emits from the API server or scheduler: manual state changes and executor-reported failures (_on_task_instance_manual_state_change). Those TIs have no TIRunContext. Today OL reads dynamic_region rows itself to get the iteration, which ties it to the table layout. The helper was meant to give it a supported call instead, returning the same data ti_run sends to the worker. With it gone, OL would do:
```
resolver = TaskCoordinateResolver.for_dag(ti.task.dag, session)
loop = resolver.get_loop_position(ti)
expanded_ti_count = resolver.get_expanded_ti_count(ti)
```
That works, but TaskCoordinateResolver is a core internal in airflow.models. Are you OK with providers relying on it for this (keeping for_dag, get_loop_position and get_expanded_ti_count stable)? Or would you prefer a different entry point, e.g. a TaskInstance method that takes the session?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323827495
##########
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:
Yes, through ti.task’s enclosing loop group. Removed the accessor and the doc row.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb merged PR #75132:
URL: https://github.com/apache/airflow/pull/75132
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4335179774
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -3266,14 +3266,51 @@
"terminal_is_mapped": {
"title": "Terminal Is Mapped",
"type": "boolean"
+ },
+ "region_id": {
+ "format": "uuid",
+ "title": "Region Id",
+ "type": "string"
+ },
+ "trajectory": {
+ "minimum": 1,
+ "title": "Trajectory",
+ "type": "integer"
+ },
+ "forked_from_region_id": {
+ "anyOf": [
Review Comment:
I went with “sent, or absent” for the new fields, using the same pattern as RegionId/RegionIndex. I did try post-processing the supervisor schema globally, but it doesn’t fit in this PR:
- replacing anyOf with oneOf, as the FastAPI app does, makes go-jsonschema emit interface{} for about 20 fields that are typed pointers today;
- collapsing nullable optional fields to “sent or absent” retypes around 80 Go fields and breaks the Go SDK build (pkg/execution, pkg/binding).
So I’d suggest doing the global fix separately, together with the Go/TS SDK side. Happy to open an issue for it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on PR #75132:
URL: https://github.com/apache/airflow/pull/75132#issuecomment-6288609487
> I'm not sure integer trajectory is right way to expose this though
The trajectory itself is identified by the loop region UUID (loop.region_id). The integer is only that region’s position in the chain of forks, the same number the task log path already shows as pass=3.2. I’d like to keep it so a single event can tell which rerun it comes from, and whether it’s newer, without having seen the other region’s UUID (UUIDv7 ordering needs both). I reworded the docs and docstrings to make that split explicit. Happy to rename it if “trajectory” clashes with something different you have in mind.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4335182350
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -5270,6 +5307,18 @@
],
"default": null
},
+ "expanded_ti_count": {
+ "anyOf": [
+ {
+ "type": "integer"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Expanded Ti Count"
+ },
Review Comment:
Agreed, done: it’s now integer without null in the schema, and the API only sends it for mapped tasks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4335520869
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -3266,14 +3266,51 @@
"terminal_is_mapped": {
"title": "Terminal Is Mapped",
"type": "boolean"
+ },
+ "region_id": {
+ "format": "uuid",
+ "title": "Region Id",
+ "type": "string"
+ },
+ "trajectory": {
+ "minimum": 1,
+ "title": "Trajectory",
+ "type": "integer"
+ },
+ "forked_from_region_id": {
+ "anyOf": [
Review Comment:
Sounds good.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #75132:
URL: https://github.com/apache/airflow/pull/75132#issuecomment-6274671847
I'll need to take another look from a computer
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323837116
##########
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:
Done: forked_from_region_id and resumes_from_index are only set for a fork, so the response leaves them out for the original region.
##########
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:
Yes, resumes_from_index is None for the original region now (and absent from the response).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4335168549
##########
go-sdk/pkg/execution/genmodels/models.gen.go:
##########
@@ -1329,11 +1333,20 @@ type LoopContext struct {
// NodeID corresponds to the JSON schema field "node_id".
NodeID string `msgpack:"node_id"`
+ // RegionID corresponds to the JSON schema field "region_id".
+ RegionID string `msgpack:"region_id"`
Review Comment:
Both now generate that way: ForkedFromRegionID *string, ResumesFromIndex *int and ExpandedTICount *int, all omitempty. The fields use the same SkipJsonSchema[None] + _drop_null_default pattern as RegionId/RegionIndex, so the schema says “this type, or absent” and the Go generator picks that up.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #75132:
URL: https://github.com/apache/airflow/pull/75132#issuecomment-6314873438
### Backport failed to create: v3-4-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/39592522472'> Run details </a>
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>v3-4-test</td>
<td><a href="https://github.com/apache/airflow/commit/e6a2c9f4e7817012cd503a8663f64aa14c2a490a"><img src='https://img.shields.io/badge/Commit-e6a2c9f-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker e6a2c9f v3-4-test
```
This should apply the commit to the v3-4-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
If you don't have cherry-picker installed, see the [installation guide](https://github.com/apache/airflow/blob/main/dev/README.md#how-to-backport-pr-with-cherry-picker-cli).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4335162779
##########
airflow-core/docs/authoring-and-scheduling/loops.rst:
##########
@@ -413,3 +431,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 previous ones from the
+iteration after the rerun gate. Each execution is a trajectory, identified by ``loop.region_id`` and
+numbered by ``loop.trajectory``: 1 for the first one, growing by one with each rerun. The rerun gate, and whatever you clear in its
+iteration, run again in the trajectory that iteration belongs to, with their next try. The
+replacement iterations run in the new trajectory, from ``loop.resumes_from_index``, and their tries
+count from the start again. In the
Review Comment:
Applied.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323835019
##########
airflow-core/src/airflow/models/task_coordinates.py:
##########
@@ -69,6 +71,40 @@ class _ProducerRequest:
loop_node_id: str | None = None
+@attrs.frozen(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
+
+
+@attrs.frozen(kw_only=True)
+class TaskInstanceExecutionPosition:
Review Comment:
No, removed it. ti_run now calls get_loop_position and get_expanded_ti_count directly.
##########
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:
Removed the helper. Server-side callers build a resolver with TaskCoordinateResolver.for_dag(dag, session), which reuses the Dag they already hold, and call the same two methods.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323868126
##########
airflow-core/src/airflow/models/task_coordinates.py:
##########
@@ -243,10 +278,47 @@ def loop_context(
regions.update(
load_region_ancestry([ti.region_id], dag_id=ti.dag_id, run_id=ti.run_id, session=self.session)
)
- position = loop_position(regions, ti.region_id, ti.region_index, group.node_id)
- if position is None:
+ coordinate = DynamicRegion.find_physical_coordinate(
+ ti.region_id, ti.region_index, group.node_id, regions
+ )
+ if coordinate is None:
raise ValueError("Task coordinates do not belong to the pinned loop")
- return group, position[1]
+ region_id, index = coordinate
+ region = regions[region_id]
+ is_fork = region.forked_from_region_id is not None
+ return LoopPosition(
+ group=group,
+ index=index,
+ region_id=region_id,
+ trajectory=get_trajectory(region, regions),
+ forked_from_region_id=region.forked_from_region_id,
+ resumes_from_index=region.resumes_from_index if is_fork else None,
+ )
+
+ def get_expanded_ti_count(
+ self, ti: TaskCoordinate, *, task: SerializedOperator | None = None
+ ) -> int | None:
+ """
+ Count the live instances of a mapped task in the expansion ``ti`` belongs to, None if not mapped.
+
+ This is what a worker gets as ``TIRunContext.expanded_ti_count``.
+ """
+ if ti.region_id == SENTINEL_REGION_ID:
+ return None
+ if task is None:
+ task = self.get_task(ti.dag_id, ti.run_id, ti.task_id, dag_version_id=ti.dag_version_id)
+ if not task.get_needs_expansion():
+ return None
+ return self.session.scalar(
+ select(func.count()).where(
Review Comment:
This adds one COUNT per mapped task instance start in ti_run. It’s covered by the task_instance_current_key index and pinned at one query in the tests. Plain and loop tasks get no new query. Fine for the hot path?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323853109
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -267,6 +268,27 @@ def model_post_init(self, context: Any) -> None:
if self.region_index is None:
self.region_index = self.map_index
+ @property
+ def loop(self) -> LoopContextAccessor | None:
+ """The iteration of the innermost loop this task instance runs in, ``None`` outside a loop."""
+ if self._ti_context_from_server is None or self._ti_context_from_server.loop is None:
+ return None
+ return LoopContextAccessor(self._ti_context_from_server.loop, self)
+
+ @property
+ def loop_decision(self) -> LoopDecision | None:
+ """A loop gate's decision, set once the gate has evaluated whether the loop stops."""
+ if self._loop_decision is None:
+ return None
+ return LoopDecision(stop=self._loop_decision.stop, reason=self._loop_decision.reason)
+
+ @property
+ def expanded_ti_count(self) -> int | None:
Review Comment:
Context already declares an expanded_ti_count key, but the Task SDK never fills it. I exposed the count on the task instance instead of filling that key, because xcom_arg.py passes the context value to get_relevant_upstream_map_indexes, so filling it would change how mapped task groups resolve upstream XComs. Inside a loop it would arguably be more correct, since GetTICount counts the latest pass rather than the caller’s iteration, but that’s a behavior change I’d rather do separately. Is keeping it on the TI the right call, or should this PR fill the context key too?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323839139
##########
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),
]
+
+
+@pytest.mark.parametrize("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:
There is, using it now.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4335173307
##########
go-sdk/pkg/execution/genmodels/models.gen.go:
##########
@@ -1329,11 +1333,20 @@ type LoopContext struct {
// NodeID corresponds to the JSON schema field "node_id".
NodeID string `msgpack:"node_id"`
+ // RegionID corresponds to the JSON schema field "region_id".
+ RegionID string `msgpack:"region_id"`
+
+ // ResumesFromIndex corresponds to the JSON schema field "resumes_from_index".
+ ResumesFromIndex interface{} `msgpack:"resumes_from_index,omitempty"`
Review Comment:
As above ^
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323831063
##########
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:
It is: the trajectory is the loop region, exposed as loop.region_id. loop.trajectory just numbers those regions along the fork chain (1 for the original). Reworded the docs to say that.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4323855013
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -267,6 +268,27 @@ def model_post_init(self, context: Any) -> None:
if self.region_index is None:
self.region_index = self.map_index
+ @property
+ def loop(self) -> LoopContextAccessor | None:
Review Comment:
Same thing as the group_id comment, one level up: this duplicates context["loop"]. Listeners can’t get it from the context cheaply: get_template_context() re-validates params on every call and can raise from a user’s template. The template context now builds its accessor from this property, so there’s one way to build it. OK to keep?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
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),
]
+
+
+@pytest.mark.parametrize("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
+@attrs.frozen(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
+
+
+@attrs.frozen(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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Expose task instance loop pass and expansion size to listeners [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #75132:
URL: https://github.com/apache/airflow/pull/75132#discussion_r4335165814
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -371,19 +371,28 @@ def ti_run(
should_retry=_is_eligible_to_retry(previous_state, ti.try_number, ti.max_tries),
multi_team=conf.getboolean("core", "multi_team"),
)
- resolver = TaskCoordinateResolver(dag_bag, session)
- if loop_context := resolver.loop_context(ti):
- group, index = loop_context
- terminal = resolver.get_task(
- ti.dag_id, ti.run_id, group.terminal_task_id, dag_version_id=ti.dag_version_id
- )
- context.loop = LoopContext(
- node_id=group.node_id,
- index=index,
- max_iterations=group.max_iterations,
- terminal_task_id=group.terminal_task_id,
- terminal_is_mapped=terminal.get_needs_expansion(),
- )
+ if ti.region_id != SENTINEL_REGION_ID:
+ resolver = TaskCoordinateResolver(dag_bag, session)
+ task = resolver.get_task(ti.dag_id, ti.run_id, ti.task_id, dag_version_id=ti.dag_version_id)
+ if (expanded_ti_count := resolver.get_expanded_ti_count(ti, task=task)) is not None:
+ context.expanded_ti_count = expanded_ti_count
+ if (loop := resolver.get_loop_position(ti, task=task)) is not None:
+ terminal = resolver.get_task(
+ ti.dag_id, ti.run_id, loop.group.terminal_task_id, dag_version_id=ti.dag_version_id
+ )
+ context.loop = LoopContext(
+ node_id=loop.group.node_id,
+ index=loop.index,
+ max_iterations=loop.group.max_iterations,
+ terminal_task_id=loop.group.terminal_task_id,
+ terminal_is_mapped=terminal.get_needs_expansion(),
+ region_id=loop.region_id,
+ trajectory=loop.trajectory,
+ )
+ # Set only for a fork, as the route leaves out unset fields
+ if loop.forked_from_region_id is not None:
+ context.loop.forked_from_region_id = loop.forked_from_region_id
+ context.loop.resumes_from_index = loop.resumes_from_index
Review Comment:
Done: TaskCoordinateResolver.get_loop_context() now builds the LoopContext the worker gets, and ti_run just assigns it. It also gives the OpenLineage provider the exact same object for server-side events, so it replaces the separate LoopPosition struct.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org