ashb opened a new pull request, #74858:
URL: https://github.com/apache/airflow/pull/74858
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 ove
rwriting 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.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull
request.
Write a good git commit message following this guide:
https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the
documentation.
For user-facing UI changes, please attach before/after screenshots (or a
short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for
a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we
sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] 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.
--
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]
