This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-2-test by this push:
new 0521aae569b [v3-2-test] Fix slow and incomplete trigger cleanup in
scheduler (#66210) (#66381)
0521aae569b is described below
commit 0521aae569bdd896b3b23a8a079de8c2a7a59052
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue May 5 09:58:50 2026 +0530
[v3-2-test] Fix slow and incomplete trigger cleanup in scheduler (#66210)
(#66381)
The periodic `_remove_unreferenced_triggers` cleanup ran on every
`parsing_cleanup_interval` and used `NOT IN` subqueries against
`task_instance`, `callback`, and `asset_watcher`. Since
`task_instance.trigger_id` and `callback.trigger_id` are nullable, a single
`NULL` row in those subqueries makes `NOT IN` evaluate to UNKNOWN for every
outer row -- so unreferenced triggers were never being deleted (in
production
nearly every `task_instance` has `trigger_id IS NULL`). The same nullability
also prevented PostgreSQL from using a hashed subplan, forcing a sequential
scan with Materialize over `task_instance` on every cycle.
Rewrite the three subqueries as `NOT EXISTS` correlated subqueries. Orphan
triggers are now reliably deleted, and PostgreSQL uses the existing
`ti_trigger_id` and `idx_awm_trigger_id` indexes to short-circuit per row.
(cherry picked from commit 1ca2797d3bbd0b6480231a9b435323c6f887ba06)
Co-authored-by: Jed Cunningham
<[email protected]>
---
airflow-core/newsfragments/66210.bugfix.rst | 1 +
airflow-core/src/airflow/jobs/scheduler_job_runner.py | 8 +++++---
airflow-core/tests/unit/jobs/test_scheduler_job.py | 18 ++++++++++++++++++
3 files changed, 24 insertions(+), 3 deletions(-)
diff --git a/airflow-core/newsfragments/66210.bugfix.rst
b/airflow-core/newsfragments/66210.bugfix.rst
new file mode 100644
index 00000000000..8d2176ac9e1
--- /dev/null
+++ b/airflow-core/newsfragments/66210.bugfix.rst
@@ -0,0 +1 @@
+Fix slow and incomplete trigger cleanup in scheduler. Unreferenced triggers
are now reliably deleted instead of accumulating without bound, and the
periodic cleanup no longer scans the full ``task_instance`` table on every
cycle.
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 2bec7843965..4e5f8289d21 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -3037,9 +3037,11 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
session.execute(
delete(Trigger)
.where(
- Trigger.id.not_in(select(AssetWatcherModel.trigger_id)),
- Trigger.id.not_in(select(Callback.trigger_id)),
- Trigger.id.not_in(select(TaskInstance.trigger_id)),
+ ~exists(
+
select(AssetWatcherModel.trigger_id).where(AssetWatcherModel.trigger_id ==
Trigger.id)
+ ),
+ ~exists(select(Callback.trigger_id).where(Callback.trigger_id
== Trigger.id)),
+
~exists(select(TaskInstance.trigger_id).where(TaskInstance.trigger_id ==
Trigger.id)),
)
.execution_options(synchronize_session="fetch")
)
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 3d717cf3817..61b7d28ba53 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -7841,6 +7841,24 @@ class TestSchedulerJob:
self.job_runner._remove_unreferenced_triggers(session=session)
assert session.scalars(select(Trigger.classpath)).one_or_none() ==
expected_classpath
+ @pytest.mark.need_serialized_dag(False)
+ def test_delete_unreferenced_triggers_with_null_trigger_id_ti(self,
dag_maker, session):
+ """Unreferenced triggers are deleted even when other TIs have
``trigger_id IS NULL``."""
+ self.job_runner = SchedulerJobRunner(job=Job())
+
+ with dag_maker(dag_id="dag", session=session):
+ EmptyOperator(task_id="task")
+ dag_maker.create_dagrun()
+
+ unreferenced = Trigger(
+
classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger",
kwargs={}
+ )
+ session.add(unreferenced)
+ session.flush()
+
+ self.job_runner._remove_unreferenced_triggers(session=session)
+ assert session.scalar(select(func.count()).select_from(Trigger)) == 0
+
@patch("airflow.serialization.serialized_objects.SerializedDAG.create_dagrun")
def test_misconfigured_dags_doesnt_crash_scheduler(self, mock_create,
session, dag_maker, caplog):
"""Test that if dagrun creation throws an exception, the scheduler
doesn't crash"""
