dabla commented on PR #62922: URL: https://github.com/apache/airflow/pull/62922#issuecomment-5511474806
> A question about how iteration state survives a retry. > > If I'm reading `XComIterable` correctly, per item results land under `return_value_0`, `return_value_1` and so on with `map_index = -1`. Does resume after failure read those keys back to skip items that already finished? > > If it does, I do not think it can work today. On every task run that is not a deferral resume, the API server collects every xcom key for the task instance and sends them to the worker as `xcom_keys_to_clear` it. And the task sdk runner deletes them on a task start up. > > For a task running for 100 items, with crash on item 60, Attempt 1's `return_value_0` through `return_value_59` are gone before attempt 2 starts, and the retry repeats all 100 items. > > A suggestion: the xcom design for results looks right to me. Downstream tasks have to read them, and that is what XCom is for. > > Its the progress record that needs a different home, and `task_state_store` from AIP-103 is built for exactly this, since rows are scoped per task instance and deliberately survive retries. The AIP-104 wiki page already mentions AIP-103 for intermediate state, so this may just be the implementation catching up with the design. > > That split might also shrink #70223. If progress lives in task state and only final results go to XCom, the bulk-key read endpoint may not be needed. @uranusjr @amoghrajesh I've completely refactored the `IterableOperator` and removed the custom retry mechanism using the defer mechanism. This version now uses the AIP-103 Task State Store to keep track of the succeeded and failed tasks, so that on each retry attempt, only the failed ones are processed again. To make the state of pushed XComs survive on each attempt of the `IterableOperator`, as by default in Airflow all XComs related to a retried task instance will be deleted before running it again, I added a special condition for the `IterableOperator` in the `run` method (now more specifically `_run_task_and_map_outcome` method) of the `task_runner` module. This way, XComs only get cleared on the first run, but not on retries. This is necessary because otherwise the XComs would get deleted and the state stored in the Task State Store wouldn't be sufficient. Yes, the succeeded tasks wouldn't be re-run unnecessarily, but the stored XComs would be gone resulting in incomplete returned results by the `XComIterable`. I chose this solution on purpose, because also storing the intermediate results in the Task State Store would be a bad idea, as those would all end up in the Airflow metadatabase. Hence, XComs are more suitable for this, as they allow custom backends to avoid this. I also added functionality on the `XComIterable` so it can be used to allow lazy fetched pagination results, I also added a third example here in the PR which shows how it can be used. There the main advantage is that you avoid keeping track of each paged result in the memory of the worker as you immediately push the paged result to the XCom backend, that way you only consume the memory of one paged result instead of accumulating those all in one list and the return the whole list as one XCom at the end of the task, which is a huge improvement. Happy to hear feedback on this, I think we are now very close to a final solution. I tested it locally with the Pokémon examples and it works. -- 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]
