Skip to content

Expose task instance loop pass and expansion size to listeners - #75132

Merged
ashb merged 5 commits into
apache:mainfrom
kacpermuda:expose-loop-and-expansion-position
Oct 21, 2026
Merged

ashb merged 5 commits into
apache:mainfrom
kacpermuda:expose-loop-and-expansion-position

Conversation

@kacpermuda

@kacpermuda kacpermuda commented Oct 19, 2026 •

Copy link
Copy Markdown
Collaborator

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:

  • 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. With this new information, we know everything from a single event.
  • 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.
  • 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. This one was not required for backtracing, but it's a really useful piece of information.

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?
  • Yes (Claude Code)

Generated-by: Claude Code (Opus 5.5) following the guidelines


  • Read the 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) is needed.
  • When adding dependency, check compliance with the ASF 3rd Party License Policy.
  • For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in 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

@ashb ashb left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure integer trajectory is right way to expose this though

@ashb

ashb commented Oct 19, 2026

Copy link
Copy Markdown
Member

I'll need to take another look from a computer

@kacpermuda

Copy link
Copy Markdown
Collaborator Author

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.

@kacpermuda
kacpermuda force-pushed the expose-loop-and-expansion-position branch from dd5a010 to 7db75fd Compare October 20, 2026 13:19
@kacpermuda kacpermuda added the full tests needed We need to run full set of tests for this PR to merge label Oct 20, 2026
@kacpermuda
kacpermuda marked this pull request as ready for review October 20, 2026 13:25

@ashb ashb left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Broadly approved, could of nits/questions about the generated schema and models

kacpermuda and others added 5 commits October 21, 2026 11:28
…ners

A single event about a task instance should be enough to tell which execution it
describes: which loop pass and rerun of the loop it ran in, how many instances its
mapped expansion has, and why a loop gate stopped. Workers only had the loop
iteration, a mapped task inside a loop could not be tied to the pass it ran in,
reruns could only be ordered by comparing region ids from two events, and server
side consumers such as listeners had to read region rows to place a stored task
instance.

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
…ed attempts

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
@kacpermuda
kacpermuda force-pushed the expose-loop-and-expansion-position branch from 2a1e812 to f3d974d Compare October 21, 2026 09:42
@ashb ashb added the backport-to-v3-4-test Backport to v3-4-test label Oct 21, 2026
@ashb
ashb merged commit e6a2c9f into apache:main Oct 21, 2026
148 checks passed
@kacpermuda
kacpermuda deleted the expose-loop-and-expansion-position branch October 21, 2026 11:32
@github-actions

Copy link
Copy Markdown
Contributor

Backport failed to create: v3-4-test. View the failure log Run details

Note: As of Merging PRs targeted for Airflow 3.X
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 Slack channel.

Status Branch Result
❌ v3-4-test Commit Link

You can attempt to backport this manually by running:

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:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 21, 2026
Uses what apache#75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 21, 2026
Uses what apache#75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 21, 2026
…eners (apache#75132)

A single event about a task instance should be enough to tell which execution it
describes: which loop pass and rerun of the loop it ran in, how many instances its
mapped expansion has, and why a loop gate stopped. Workers only had the loop
iteration, a mapped task inside a loop could not be tied to the pass it ran in,
reruns could only be ordered by comparing region ids from two events, and server
side consumers such as listeners had to read region rows to place a stored task
instance.
(cherry picked from commit e6a2c9f)

Co-authored-by: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 21, 2026
…eners (apache#75132)

A single event about a task instance should be enough to tell which execution it
describes: which loop pass and rerun of the loop it ran in, how many instances its
mapped expansion has, and why a loop gate stopped. Workers only had the loop
iteration, a mapped task inside a loop could not be tied to the pass it ran in,
reruns could only be ordered by comparing region ids from two events, and server
side consumers such as listeners had to read region rows to place a stored task
instance.
(cherry picked from commit e6a2c9f)

Co-authored-by: Claude <noreply@anthropic.com>
ashb pushed a commit that referenced this pull request Oct 21, 2026
…eners (#75132) (#75309)

A single event about a task instance should be enough to tell which execution it
describes: which loop pass and rerun of the loop it ran in, how many instances its
mapped expansion has, and why a loop gate stopped. Workers only had the loop
iteration, a mapped task inside a loop could not be tied to the pass it ran in,
reruns could only be ordered by comparing region ids from two events, and server
side consumers such as listeners had to read region rows to place a stored task
instance.
(cherry picked from commit e6a2c9f)

Co-authored-by: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 21, 2026
Uses what apache#75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 21, 2026
…code

The API server holds task instances without their task when a user sets a
state, so those events never carried the airflow run facet the guide promised.
The fallbacks for loop details only served Airflow builds from before
apache#75132, which no release contains. Timing the stored position
lookup makes its per task instance queries in the scheduler visible.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 22, 2026
Uses what apache#75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 22, 2026
…code

The API server holds task instances without their task when a user sets a
state, so those events never carried the airflow run facet the guide promised.
The fallbacks for loop details only served Airflow builds from before
apache#75132, which no release contains. Timing the stored position
lookup makes its per task instance queries in the scheduler visible.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 22, 2026
Uses what apache#75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 22, 2026
…code

The API server holds task instances without their task when a user sets a
state, so those events never carried the airflow run facet the guide promised.
The fallbacks for loop details only served Airflow builds from before
apache#75132, which no release contains. Timing the stored position
lookup makes its per task instance queries in the scheduler visible.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 23, 2026
Uses what apache#75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 23, 2026
…code

The API server holds task instances without their task when a user sets a
state, so those events never carried the airflow run facet the guide promised.
The fallbacks for loop details only served Airflow builds from before
apache#75132, which no release contains. Timing the stored position
lookup makes its per task instance queries in the scheduler visible.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 23, 2026
Uses what apache#75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>
kacpermuda added a commit to kacpermuda/airflow that referenced this pull request Oct 23, 2026
…code

The API server holds task instances without their task when a user sets a
state, so those events never carried the airflow run facet the guide promised.
The fallbacks for loop details only served Airflow builds from before
apache#75132, which no release contains. Timing the stored position
lookup makes its per task instance queries in the scheduler visible.

Co-Authored-By: Claude <noreply@anthropic.com>
mobuchowski pushed a commit that referenced this pull request Oct 26, 2026
* Improve task loop information and tests for OpenLineage provider

* Fix mypy and docs spelling errors in OpenLineage task loop changes

Provider mypy and the docs spellcheck run in CI through breeze rather than
the default prek hooks, so they were not caught locally before the previous push.

Co-Authored-By: Claude <noreply@anthropic.com>

* Expose loop gate decisions in OpenLineage task instance facet

Alerting on why a loop stopped (for example, its until condition still being false at
max_iterations) needs the gate's decision in the event itself, without reading the Airflow
metadata database.

Co-Authored-By: Claude <noreply@anthropic.com>

* Keep OpenLineage events when the loop iteration can't be read

The run id must not depend on the optional loop iteration, and the lineage macro should
place a task instance the same way the listener does.

Co-Authored-By: Claude <noreply@anthropic.com>

* Address review of OpenLineage task loop changes

A region of an unexpected shape, or a failed lookup of a mapped task's iteration, must not
merge the runs of loop iterations or cost an event. Tests now pin the region shapes Airflow
creates, so a change to them fails the provider's tests instead of silently changing events.

Co-Authored-By: Claude <noreply@anthropic.com>

* Stub lineage macros instead of OpenLineage internals in provider tests

The dbt Cloud and Dataproc tests patched how OpenLineage builds a task's run id, so they broke
when that changed. Stubbing the macros they actually use keeps them independent of it.

Co-Authored-By: Claude <noreply@anthropic.com>

* Describe loop reruns and expansion size in OpenLineage task events

Uses what #75132 exposes, so a single event tells which rerun of a
loop it comes from, whether that rerun superseded an earlier one, and how many
mapped instances to expect, and events for manual state changes get the same
details as worker events. Loop details are grouped in one object so it can
describe enclosing loops if Airflow ever nests them, and iterations are counted
from one as in Airflow.

Co-Authored-By: Claude <noreply@anthropic.com>

* Fix OpenLineage manual state change events on Airflow before 3.4

Stored task instances have no region_id before Airflow 3.4, so reading it before
checking the version failed the event.

Co-Authored-By: Claude <noreply@anthropic.com>

* Clarify OpenLineage events for UI state changes and drop pre-release code

The API server holds task instances without their task when a user sets a
state, so those events never carried the airflow run facet the guide promised.
The fallbacks for loop details only served Airflow builds from before
#75132, which no release contains. Timing the stored position
lookup makes its per task instance queries in the scheduler visible.

Co-Authored-By: Claude <noreply@anthropic.com>

* Avoid reloading Dags for OpenLineage events of failed task instances

Placing a task instance pinned to another Dag version than the one the
scheduler holds made Airflow load and deserialize that version again for
every such task instance, inside the scheduler loop. Those now fall back to
reading their region rows. That bounds the cost instead of only measuring
it, so the timer added for it goes away before any release has it.

Co-Authored-By: Claude <noreply@anthropic.com>

* Remove the OpenLineage task execution guide from this change

The guide will be written separately, so this change only keeps the facet
schema as the description of the new fields.

Co-Authored-By: Claude <noreply@anthropic.com>

* Describe task state, Dag version and loop once in OpenLineage task events

Consumers placing a task event need its state to tell a retry from a final
failure, and the Dag version it ran, from the event alone. They also need the
same fields whether the worker or the scheduler sent it, and each fact in one
place: a loop's definition that also showed in the task's own group read as
two things that could disagree.

Co-Authored-By: Claude <noreply@anthropic.com>

* Describe mapping and run conditions of tasks in the OpenLineage job facet

Consumers that prepare for a Dag run's task events, for example to alert on a
task group, need to know before the tasks run which of them expand, which
groups are mapped or loops, and what lets a task run or keeps it from running.

Co-Authored-By: Claude <noreply@anthropic.com>

* Expect no Dag version from Airflow 3.0 runtime task instances in OpenLineage tests

The provider's compatibility tests against Airflow 3.0 failed, as its runtime
task instances don't carry a Dag version, which the event then leaves out.

Co-Authored-By: Claude <noreply@anthropic.com>

* Cover every task group and loop arrangement in OpenLineage system tests

Task events place a task by its own group, the loop around it and its
iteration, which differ by where the task sits: in a plain or nested group, in
a mapped group, in a group inside a loop, or in a loop that is itself in a
group. Each arrangement now has a task whose events, and the Dag's job facet,
the e2e run checks.

Co-Authored-By: Claude <noreply@anthropic.com>

* Send OpenLineage events when a task instance's Dag version can't be read

Reading a stored task instance's Dag version can query the database, and a
failure there dropped the whole event the scheduler sent about it. The bundle
name and version only describe the event, so it now goes out without them.

Co-Authored-By: Claude <noreply@anthropic.com>

* Simplify OpenLineage facet schemas and check more of each event in system tests

The facet schemas only describe fields loosely, as they guarantee none, so
they now add plain types. Loop groups tell independently whether they're
mapped, as a loop may become mappable, and fields read from Airflow objects
tolerate versions that lack them. System tests check every field of the new
events, per Airflow version where it differs, instead of skipping optional
ones, and Airflow 2 expects a literal expansion's size.

Co-Authored-By: Claude <noreply@anthropic.com>

* Name the OpenLineage task group id after what it is

The id Airflow identifies a group by is its own id prefixed by its parents'
only where they prefix their children, so it isn't always a full path.

Co-Authored-By: Claude <noreply@anthropic.com>

* Expect a literal expansion's size from Task SDK operators before Airflow 3.2

The provider's compatibility tests against Airflow 3.0 and 3.1 failed, as their
Task SDK operators still count a literal expansion before serialization.

Co-Authored-By: Claude <noreply@anthropic.com>

* Expect no COMPLETE for a skipped task on Airflow 3.0 in OpenLineage e2e tests

Airflow 3.0 calls no listener when a task skips itself, so the provider can't
emit the event there, which failed the e2e compatibility test.

Co-Authored-By: Claude <noreply@anthropic.com>

* Leave a skipped task's START unchecked on Airflow 3.0 in OpenLineage e2e tests

Without a COMPLETE, the START alone fails the runId check on Airflow 3.0, where
no listener runs for a task that skips itself.

Co-Authored-By: Claude <noreply@anthropic.com>

* Build scheduler-sent OpenLineage task events without extra queries

Reading the loop context, Dag version and expansion size inside the
scheduler's failure handling flushed its pending changes through savepoints
and cost several queries per failed task instance. A Dag edited after an
attempt started could also give its FAIL another runId than its START, the
stored log_url pointed a loop iteration at a mapped index, and before
Airflow 3.4 the attempt already held its next id when the event was built.
The system tests could not notice a disabled runId check, loop fields that
contradict each other, or a checking task that retries and hides the first
mismatch.

Co-Authored-By: Claude <noreply@anthropic.com>

* Leave the task instance note out of OpenLineage task events

The note was meant to show manual state changes, but on Airflow 3 their
events carry no airflow facet, so it never reached them. Only the FAIL the
scheduler sends itself read it, at the cost of a query on the scheduler's
session for every task instance it fails, and Airflow 2 workers read it with
a query for every task event.

Co-Authored-By: Claude <noreply@anthropic.com>

* Leave the log link of OpenLineage task events to Airflow

Airflow 3.4.0, the first release with task loops, links a loop's task
instances to their own page (#75458), so rewriting the link in the provider
would only have patched the 3.4.0 pre-releases.

Co-Authored-By: Claude <noreply@anthropic.com>

* Describe what the region coordinates of OpenLineage task events address

Consumers could not tell from the schema why an event carries both region
fields, though together they are what Airflow's REST API and UI select a
task instance in a loop or expansion by.

Co-Authored-By: Claude <noreply@anthropic.com>

* Send the FAIL the scheduler emits for a mapped task to OpenLineage

On the scheduler a mapped task is a serialized mapped operator, which leaves
the documentation fields its task doesn't set undefined. Reading them failed
the event, so a mapped task instance the scheduler failed, e.g. after its
worker died, never got a FAIL to close its run.

Co-Authored-By: Claude <noreply@anthropic.com>

* Share what OpenLineage system test Dags repeated

The loop Dags defined the same event order twice, and five Dag docstrings
repeated what a task event's identity covers. A test also promised to check
the loop's max_iterations, which it never did.

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:API Airflow's REST/HTTP API area:go-sdk area:logging area:task-sdk area:ts-sdk backport-to-v3-4-test Backport to v3-4-test full tests needed We need to run full set of tests for this PR to merge kind:documentation

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants