Andrushika commented on code in PR #70128: URL: https://github.com/apache/airflow/pull/70128#discussion_r3658572511
########## airflow-core/src/airflow/assets/manager.py: ########## @@ -65,6 +65,18 @@ log = structlog.get_logger(__name__) +def _sorted_by_dag_id(dags: Collection[DagModel]) -> list[DagModel]: + """ + Order dags deterministically before queueing their AssetDagRunQueue rows. + + ``dags_to_queue`` is a set, whose iteration order varies between processes. If two + concurrent transactions queue the same dags for one asset in different orders, they take + the per-row locks (from ON CONFLICT / ON DUPLICATE KEY / the SAVEPOINT merge) in opposite + orders and can deadlock. Inserting in a fixed order removes that lock-ordering cycle. + """ + return sorted(dags, key=lambda dag: dag.dag_id) Review Comment: We have one in the SDK DAG class(`__lt__` comparing `dag_id`, in `task-sdk/src/airflow/sdk/definitions/dag.py`), but the ORM DagModel doesn't have `__lt__`. The objects sorted here are ORM DagModel rows. So I think keeping the explicit key here is better than adding `__lt__` inside this fix... Maybe we don't need it since this seems to be the only place we use it. (Happy to do that as a follow-up if you think it is useful!) -- 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]
