This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 2c1035a9732 Speed up Trigger.clean_unused query to prevent Triggerer
crashes (#68244)
2c1035a9732 is described below
commit 2c1035a973227a1dd5187429917ff2d60afee774
Author: AntonioBergonzi <[email protected]>
AuthorDate: Wed Jul 29 10:38:58 2026 +0200
Speed up Trigger.clean_unused query to prevent Triggerer crashes (#68244)
* Use NOT EXISTS anti-join in Trigger.clean_unused instead of LEFT JOIN +
aggregate
* add FOR UPDATE SKIP LOCKED to Trigger.clean_unused to avoid deadlocks
between concurrent triggerer pods
* use with_row_locks helper and scope lock to trigger table
---
airflow-core/src/airflow/models/trigger.py | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
diff --git a/airflow-core/src/airflow/models/trigger.py
b/airflow-core/src/airflow/models/trigger.py
index 386a6c931fc..ef2adefd2eb 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -250,13 +250,12 @@ class Trigger(Base):
)
# Get all triggers that have no task instances, assets, or callbacks
depending on them and delete them
- ids = (
- select(cls.id)
- .where(~cls.assets.any(), ~cls.callback.has())
- .join(TaskInstance, cls.id == TaskInstance.trigger_id,
isouter=True)
- .group_by(cls.id)
- .having(func.count(TaskInstance.trigger_id) == 0)
+ ids = select(cls.id).where(
+ ~cls.assets.any(),
+ ~cls.callback.has(),
+ ~cls.task_instance.has(),
)
+ ids = with_row_locks(ids, session, of=cls, skip_locked=True,
key_share=False)
if get_dialect_name(session) == "mysql":
# MySQL doesn't support DELETE with JOIN, so we need to do it in
two steps
ids_list = list(session.scalars(ids).all())
