mirror of
https://github.com/apache/airflow.git
synced 2026-11-04 07:12:48 +00:00
Ash Berlin-Taylor
a27f8e74da
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.