Repository navigation
Let XCom backends tell loop iterations apart through store_value - #74858
Merged
Merged
Conversation
ashb
requested review from
XD-DENG,
ephraimbuddy,
hussein-awala and
jedcunningham
as code owners
October 16, 2026 14:35
ashb
force-pushed
the
xcom-loop-coordinates
branch
from
October 16, 2026 14:36
3ba2acb to
1d538ca
Compare
ashb
force-pushed
the
xcom-loop-coordinates
branch
from
October 16, 2026 16:49
1d538ca to
da27c6d
Compare
kaxil
reviewed
Oct 17, 2026
kaxil
approved these changes
Oct 17, 2026
Every iteration of a task inside a loop is its own task instance, but an XCom backend's `serialize_value` only receives dag_id, run_id, task_id, map_index and key, which is shared by different iterations of the same task. A backend might naievly derives its storage path from those values would every iteration to the same location, so every read would give back the last iteration's value n times. It only shows when the XComs are read from outside the loop, which makes it easy to miss. To work around this, without breaking old XCom backends, I have introduced a new method on the backend, `store_value`, and instead of 4 (or 5 or 6) arguments it takes a single arg of type `XComCoordinate` to prevent arg sprawl and to make this new fn future proof. `XCom.set` on another task (i.e. passign a different `task_id`) cannot know which task instance the server will resolve its coordinates to, so there the id is unknown and the location has to be unique to the write. Leaving that distinction to each backend would make the collision easy to reintroduce, so `XComCoordinate.build_path` builds the location for both cases and shortens the readable parts only when the path would exceed the backend-provided key limit. `serialize_value` already combined converting a value with writing it somewhere, and one more keyword would have kept growing its signature. For each old and new hook fn, the override closest to the class wins, so a subclass of a store_value backend that overrides serialize_value keeps its override. Airflow can't tell whether a legacy backend's locations are unique, so implementing store_value is what marks a backend as safe in a loop: a Dag that uses a loop fails at startup under a serialize_value-only backend instead of overwriting results silently. Writes and deletes of another task's XCom did not follow the rule reads use: they resolved to the caller's own task instance or the single live one, and failed with 409 when several iterations of a loop task were live. Resolving them like reads, from the calling task instance, lets a task in a loop write its own iteration of a sibling and gives a caller outside the loop the same 400 a read gets. The backend's store and load talk to external storage, so the async paths, which share an event loop with the triggerer's other triggers, run them in a thread. The trigger runner already carried its task instance id in a context variable so the API could resolve a trigger's requests. It moved into the Task SDK and holds the task instance, because the SDK has to recognise when a trigger writes its own coordinates and then pass the id and region on. The common.io ObjectStore backend now builds its object paths with build_path, so they now sit under `ti-<id>` or `write-<rand_uuid>`. Rows already written keep working because they store full paths in the metadata, and read just reads the path it is given. Overwritten and failed writes still leave their old objects behind, as they do today; cleaning those up is separate work. The configured backend also applies only to the XComs that Python tasks and triggers write and read. Tasks written in other languages still store theirs directly in the database, and fixing that needs the supervisor to apply the backend for them. Again, this doesn't change from today.
ashb
force-pushed
the
xcom-loop-coordinates
branch
from
October 18, 2026 15:20
da27c6d to
c16d7be
Compare
Contributor
|
Hi maintainer, this PR was merged without a milestone set.
|
Contributor
Backport successfully created: v3-4-testNote: As of Merging PRs targeted for Airflow 3.X In matter of doubt please ask in #release-management Slack channel.
|
ashb
added a commit
that referenced
this pull request
Oct 18, 2026
…e_value (#74858) (#75011) Every iteration of a task inside a loop is its own task instance, but an XCom backend's `serialize_value` only receives dag_id, run_id, task_id, map_index and key, which is shared by different iterations of the same task. A backend might naievly derives its storage path from those values would every iteration to the same location, so every read would give back the last iteration's value n times. It only shows when the XComs are read from outside the loop, which makes it easy to miss. To work around this, without breaking old XCom backends, I have introduced a new method on the backend, `store_value`, and instead of 4 (or 5 or 6) arguments it takes a single arg of type `XComCoordinate` to prevent arg sprawl and to make this new fn future proof. `XCom.set` on another task (i.e. passign a different `task_id`) cannot know which task instance the server will resolve its coordinates to, so there the id is unknown and the location has to be unique to the write. Leaving that distinction to each backend would make the collision easy to reintroduce, so `XComCoordinate.build_path` builds the location for both cases and shortens the readable parts only when the path would exceed the backend-provided key limit. `serialize_value` already combined converting a value with writing it somewhere, and one more keyword would have kept growing its signature. For each old and new hook fn, the override closest to the class wins, so a subclass of a store_value backend that overrides serialize_value keeps its override. Airflow can't tell whether a legacy backend's locations are unique, so implementing store_value is what marks a backend as safe in a loop: a Dag that uses a loop fails at startup under a serialize_value-only backend instead of overwriting results silently. Writes and deletes of another task's XCom did not follow the rule reads use: they resolved to the caller's own task instance or the single live one, and failed with 409 when several iterations of a loop task were live. Resolving them like reads, from the calling task instance, lets a task in a loop write its own iteration of a sibling and gives a caller outside the loop the same 400 a read gets. The backend's store and load talk to external storage, so the async paths, which share an event loop with the triggerer's other triggers, run them in a thread. The trigger runner already carried its task instance id in a context variable so the API could resolve a trigger's requests. It moved into the Task SDK and holds the task instance, because the SDK has to recognise when a trigger writes its own coordinates and then pass the id and region on. The common.io ObjectStore backend now builds its object paths with build_path, so they now sit under `ti-<id>` or `write-<rand_uuid>`. Rows already written keep working because they store full paths in the metadata, and read just reads the path it is given. Overwritten and failed writes still leave their old objects behind, as they do today; cleaning those up is separate work. The configured backend also applies only to the XComs that Python tasks and triggers write and read. Tasks written in other languages still store theirs directly in the database, and fixing that needs the supervisor to apply the backend for them. Again, this doesn't change from today. (cherry picked from commit a27f8e7) Co-authored-by: Ash Berlin-Taylor <ash@astronomer.io> Co-authored-by: Rahul Vats <43964496+vatsrahul1001@users.noreply.github.com>
1 task done
1 task done
vatsrahul1001
added a commit
that referenced
this pull request
Oct 20, 2026
…e_value (#74858) (#75011) Every iteration of a task inside a loop is its own task instance, but an XCom backend's `serialize_value` only receives dag_id, run_id, task_id, map_index and key, which is shared by different iterations of the same task. A backend might naievly derives its storage path from those values would every iteration to the same location, so every read would give back the last iteration's value n times. It only shows when the XComs are read from outside the loop, which makes it easy to miss. To work around this, without breaking old XCom backends, I have introduced a new method on the backend, `store_value`, and instead of 4 (or 5 or 6) arguments it takes a single arg of type `XComCoordinate` to prevent arg sprawl and to make this new fn future proof. `XCom.set` on another task (i.e. passign a different `task_id`) cannot know which task instance the server will resolve its coordinates to, so there the id is unknown and the location has to be unique to the write. Leaving that distinction to each backend would make the collision easy to reintroduce, so `XComCoordinate.build_path` builds the location for both cases and shortens the readable parts only when the path would exceed the backend-provided key limit. `serialize_value` already combined converting a value with writing it somewhere, and one more keyword would have kept growing its signature. For each old and new hook fn, the override closest to the class wins, so a subclass of a store_value backend that overrides serialize_value keeps its override. Airflow can't tell whether a legacy backend's locations are unique, so implementing store_value is what marks a backend as safe in a loop: a Dag that uses a loop fails at startup under a serialize_value-only backend instead of overwriting results silently. Writes and deletes of another task's XCom did not follow the rule reads use: they resolved to the caller's own task instance or the single live one, and failed with 409 when several iterations of a loop task were live. Resolving them like reads, from the calling task instance, lets a task in a loop write its own iteration of a sibling and gives a caller outside the loop the same 400 a read gets. The backend's store and load talk to external storage, so the async paths, which share an event loop with the triggerer's other triggers, run them in a thread. The trigger runner already carried its task instance id in a context variable so the API could resolve a trigger's requests. It moved into the Task SDK and holds the task instance, because the SDK has to recognise when a trigger writes its own coordinates and then pass the id and region on. The common.io ObjectStore backend now builds its object paths with build_path, so they now sit under `ti-<id>` or `write-<rand_uuid>`. Rows already written keep working because they store full paths in the metadata, and read just reads the path it is given. Overwritten and failed writes still leave their old objects behind, as they do today; cleaning those up is separate work. The configured backend also applies only to the XComs that Python tasks and triggers write and read. Tasks written in other languages still store theirs directly in the database, and fixing that needs the supervisor to apply the backend for them. Again, this doesn't change from today. (cherry picked from commit a27f8e7) Co-authored-by: Ash Berlin-Taylor <ash@astronomer.io> Co-authored-by: Rahul Vats <43964496+vatsrahul1001@users.noreply.github.com>
28 of 93 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Every iteration of a task inside a loop is its own task instance, but an XCom backend's
serialize_valueonly receives dag_id, run_id, task_id, map_index and key, which is shared by different iterations of the same task. A backend might naievly derives its storage path from those values would every iteration to the same location, so every read would give back the last iteration's value n times. It only shows when the XComs are read from outside the loop, which makes it easy to miss.To work around this, without breaking old XCom backends, I have introduced a new method on the backend,
store_value, and instead of 4 (or 5 or 6) arguments it takes a single arg of typeXComCoordinateto prevent arg sprawl and to make this new fn future proof.XCom.seton another task (i.e. passign a differenttask_id) cannot know which task instance the server will resolve its coordinates to, so there the id is unknown and the location has to be unique to the write. Leaving that distinction to each backend would make the collision easy to reintroduce, soXComCoordinate.build_pathbuilds the location for both cases and shortens the readable parts only when the path would exceed the backend-provided key limit.serialize_valuealready combined converting a value with writing it somewhere, and one more keyword would have kept growing its signature. For each old and new hook fn, the override closest to the class wins, so a subclass of a store_value backend that overrides serialize_value keeps its override. Airflow can't tell whether a legacy backend's locations are unique, so implementing store_value is what marks a backend as safe in a loop: a Dag that uses a loop fails at startup under a serialize_value-only backend instead of overwriting results silently.Writes and deletes of another task's XCom did not follow the rule reads
use: they resolved to the caller's own task instance or the single live
one, and failed with 409 when several iterations of a loop task were live.
Resolving them like reads, from the calling task instance, lets a task in
a loop write its own iteration of a sibling and gives a caller outside the
loop the same 400 a read gets.
The backend's store and load talk to external storage, so the async paths,
which share an event loop with the triggerer's other triggers, run them in
a thread.
The trigger runner already carried its task instance id in a context
variable so the API could resolve a trigger's requests. It moved into the
Task SDK and holds the task instance, because the SDK has to recognise when
a trigger writes its own coordinates and then pass the id and region on.
The common.io ObjectStore backend now builds its object paths with build_path, so
they now sit under
ti-<id>orwrite-<rand_uuid>. Rows already written keepworking because they store full paths in the metadata, and read just reads the
path it is given.
Overwritten and failed writes still leave their old objects behind, as they
do today; cleaning those up is separate work. The configured backend also
applies only to the XComs that Python tasks and triggers write and read.
Tasks written in other languages still store theirs directly in the
database, and fixing that needs the supervisor to apply the backend for
them. Again, this doesn't change from today.
Was generative AI tooling used to co-author this PR?
{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.