dabla commented on PR #62922: URL: https://github.com/apache/airflow/pull/62922#issuecomment-6341283380
@kaxil @amoghrajesh @ashb a design question before I go any further, as it touches the runner and the execution API rather than the operator. What I measured. I measured on a retry of an iterated task over ~16,000 items, about 15,400 of which had succeeded on the previous attempt. Since the checkpoints are written a round at a time, reading them back takes seconds. What's left of a retry is the XComs: Before execute, the runner deletes every key in xcom_keys_to_clear ([task_runner.py](https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/execution_time/task_runner.py), the loop over xcom_keys_to_clear). For an iterated task those are the return_value_<index> of every index, and XCom.delete first reads each value so a custom backend can purge it. That's a GET and a DELETE per index: ~29 minutes for 15,400 keys at ~117 ms per request. The indices that already succeeded are then replayed from their checkpoints, and since their XComs are gone, each result is pushed again, one POST per index: ~45 minutes at ~165 ms per request. So an index that had already succeeded costs three requests per retry, to end up with the same XCom it started with. Nothing else is left in the retry: the task process sends its requests one at a time, and these fill the channel. The question: would you accept that, for an iterated task, the runner doesn't delete the XComs up front, and the operator removes only those whose index ends differently? For example: (a) the API server leaves an iterated task's return_value_<index> keys out of xcom_keys_to_clear (it knows the task is an IterableOperator from the serialized Dag), or (b) the runner skips the up-front clear when the task says it handles its own XComs, the same kind of opt-out as enforces_execution_timeout() for the indexed tasks. The operator would then delete per index, only where the outcome changed: an index that runs again and now returns None (it pushes nothing, so its old value would be read instead of None); an index whose checkpoint is not trusted (from before a clear, or for a different input), which is rerun and overwritten, or deleted if it no longer pushes; an index that succeeded before and is now skipped or fails; after a clear that shortens the input, the keys past the new length (XComIterable never reads them, but they would never be cleaned up either). Every other key is either replayed with exactly the value it holds, or overwritten when its index runs again. On the retry above, that removes both the 29 minutes and the 45 minutes. The alternative that keeps the runner as it is would be a bulk XCom write (and ideally a bulk delete) in the execution API: still two full passes over every key per retry, but in ~100 requests instead of ~46,000. I'd rather not add that if (a) or (b) is acceptable. Which would you prefer, or do you see a reason the up-front clear has to stay for iterated tasks? This is something that has not to be addressed in this PR, any solution could be done in a follow up PR, but I already wanted to ask what would be acceptable. -- 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]
