[v3-1-test] Fix dag-processor crashing due to MySql deadlock errors (#60166)

MySQL may throw deadlock errors when multiple DAG-Processor instances are
running. The issue is a fetch sub-query being used within a delete query, which
is sometimes causing a deadlock in MySQL.

---------
(cherry picked from commit dfcd049)

Co-authored-by: Sumit Maheshwari <msumit@users.noreply.github.com>
Co-authored-by: Sumit Maheshwari <sumitm@uber.com>
Co-authored-by: Jarek Potiuk <jarek@potiuk.com>

Author: 3 people

airflow-core/src/airflow/models/asset.py

@@ -108,12 +108,17 @@ def remove_references_to_deleted_dags(session: Session):
         DagScheduleAssetAliasReference,
         TaskOutletAssetReference,
     ]
-    for model in models_to_check:
-        session.execute(
-            delete(model)
-            .where(model.dag_id.in_(select(DagModel.dag_id).where(DagModel.is_stale)))
-            .execution_options(synchronize_session="fetch")
-        )
+
+    # The queries need to be done in separate steps, because in the case of multiple
+    # dag processors on MySQL, there could be a deadlock caused by acquiring both an
+    # exclusive lock for deletion and shared lock for query in reverse sequence
+    if stale_dag_ids := session.scalars(select(DagModel.dag_id).where(DagModel.is_stale)).all():
+        for model in models_to_check:
+            session.execute(
+                delete(model)
+                .where(model.dag_id.in_(stale_dag_ids))
+                .execution_options(synchronize_session="fetch")
+            )
 
 
 alias_association_table = Table(