dabla commented on PR #62922:
URL: https://github.com/apache/airflow/pull/62922#issuecomment-6533234377
Answering my own question above, since `main` has moved under it.
**The up-front clear no longer happens for a retry.** Since #74222 ("Keep
retried task attempts and their data under the attempt UUID", in 3.4) every
attempt is its own `task_instance` row and XComs belong to the attempt that
pushed them (`xcom_v2.task_instance_id`, [cascade on the
attempt](https://github.com/apache/airflow/blob/1c3b3def8694cbd96bad9f82af80d532386e1f04/airflow-core/src/airflow/models/xcom.py#L137)).
`ti_run` builds `xcom_keys_to_clear` from [the running attempt's own
id](https://github.com/apache/airflow/blob/1c3b3def8694cbd96bad9f82af80d532386e1f04/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py#L356),
which owns nothing yet on a retry, so [the runner's delete
loop](https://github.com/apache/airflow/blob/1c3b3def8694cbd96bad9f82af80d532386e1f04/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L1653)
has nothing to do. Checked on a 3.4.0rc1 image: three attempts of one iterated
task each kept their `return_value_<index>` rows unde
r their own attempt id, nothing was deleted. So options (a) and (b) are moot,
and the ~29 minutes of GET + DELETE per index are gone on `main`.
**What is left is the replay.** The previous attempt's rows exist, but
readers resolve the live attempt: the ORM hides superseded attempts unless a
query opts in with
[`include_all_attempts`](https://github.com/apache/airflow/blob/1c3b3def8694cbd96bad9f82af80d532386e1f04/airflow-core/src/airflow/models/xcom.py#L399),
and neither `xcom_pull` nor the execution API's XCom read exposes that. So a
retry still has to push every succeeded index's value again under its own
attempt, one POST per index (~45 minutes at ~16,000 items), and the checkpoint
has to carry the value to be able to do so. The value is therefore stored
twice, once in XCom and once in the task state store, which is also why the
docs of this PR advise an object storage backend for the state store.
**Proposal for a follow-up PR, not this one.** Let the retry take over the
rows the previous attempt already wrote, and keep the checkpoint to what has no
XCom row:
- An execution API call, "promote the XComs of this task instance's
superseded attempt to the running attempt", returning the keys it promoted.
Server-side it is one statement over `xcom_v2`; whether it copies the rows (the
superseded attempt keeps its XComs visible in the history, as #74222 intends)
or moves them is the one design choice I'd like your view on, @ashb.
- The operator calls it once at the start of a retry. An index whose key
came back is done and is not run again; one whose key is missing runs again,
which also covers an attempt killed between the push and the write of its
checkpoint round. The checkpoint keeps the status, the input fingerprint and
the outlet events, and drops `result` and the other pushed XComs: the XCom row
is the proof of completion, the checkpoint the hint that saves the lookup.
Nothing changes for downstream readers.
| | this PR | with the promote call |
|---|---|---|
| first attempt, per index | one XCom push + its share of a checkpoint round
carrying the value | one XCom push + its share of a round carrying a status |
| state store | every value twice (XCom and state store) | a few hundred
bytes per round |
| retry, succeeded indices | read the rounds + one push per index | read the
rounds + one call |
Does a promote call like this fit what you have in mind for clearing midway
through an AIP-111 loop, where the same question arises for the iterations that
are kept? If so I'd open it as its own PR once this one is in: the route and
SDK message first, then the operator change.
--
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]
