This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new bb8198f6cab [v3-3-test] Speed up Trigger.clean_unused query to prevent
Triggerer crashes (#68244) (#70668)
bb8198f6cab is described below
commit bb8198f6cab5b3d2c971c4b0c77cd0724ca193c0
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jul 30 18:50:51 2026 +0200
[v3-3-test] Speed up Trigger.clean_unused query to prevent Triggerer
crashes (#68244) (#70668)
* 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
(cherry picked from commit 2c1035a973227a1dd5187429917ff2d60afee774)
Co-authored-by: AntonioBergonzi
<[email protected]>
Co-authored-by: Rahul Vats <[email protected]>
---
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 14473f36fda..a7e7b8ee293 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())
