Batch the scheduler's unreferenced-trigger cleanup deletes

airflow-core/src/airflow/config_templates/config.yml

@@ -2780,6 +2780,18 @@ scheduler:
       type: integer
       example: ~
       default: "60"
+    unreferenced_triggers_cleanup_batch_size:
+      description: |
+        Maximum number of unreferenced ``trigger`` rows the scheduler removes per
+        transaction during its periodic trigger cleanup (see
+        ``[scheduler] parsing_cleanup_interval``). Batching avoids holding row locks
+        on the ``trigger`` table for the duration of a single large delete, which
+        would otherwise block the triggerer's own writers and stall the scheduler
+        main loop. Must be at least ``1``.
+      version_added: 3.3.0
+      type: integer
+      example: ~
+      default: "500"
     dag_cache_size:
       description: |
         Max number of deserialized SerializedDAG objects the scheduler keeps in memory, keyed by Dag version

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -85,7 +85,6 @@
     AssetEvent,
     AssetModel,
     AssetPartitionDagRun,
-    AssetWatcherModel,
     DagScheduleAssetAliasReference,
     DagScheduleAssetReference,
     PartitionedAssetKeyLog,
@@ -4193,18 +4192,22 @@ def _generate_task_instance_heartbeat_timeout_message_details(
 
     @provide_session
     def _remove_unreferenced_triggers(self, *, session: Session = NEW_SESSION) -> None:
-        """Remove triggers that are no longer used by anything."""
-        session.execute(
-            delete(Trigger)
-            .where(
-                ~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")
-        )
+        """
+        Remove triggers that are no longer used by anything.
+
+        Delegates to ``Trigger._delete_unused_batch`` so the same LIMIT/row-lock/commit
+        pattern the triggerer uses (#70961) applies here too — holding row locks on
+        ``trigger`` for the duration of an unbounded DELETE stalls the triggerer's
+        writers and the scheduler main loop on installs with large trigger tables.
+
+        Batch size is controlled by ``[scheduler] unreferenced_triggers_cleanup_batch_size``
+        (default 500). One batch is removed per call; later scheduler ticks drain the
+        remainder, matching the triggerer's per-tick cadence.
+        """
+        batch_size = conf.getint("scheduler", "unreferenced_triggers_cleanup_batch_size", fallback=500)
+        if batch_size <= 0:
+            raise ValueError("[scheduler] unreferenced_triggers_cleanup_batch_size must be at least 1")
+        Trigger._delete_unused_batch(batch_size, session=session)
 
     @provide_session
     def _update_asset_orphanage(self, *, session: Session = NEW_SESSION) -> None:

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -10279,6 +10279,100 @@ def test_delete_unreferenced_triggers_with_null_trigger_id_ti(self, dag_maker, s
         self.job_runner._remove_unreferenced_triggers(session=session)
         assert session.scalar(select(func.count()).select_from(Trigger)) == 0
 
+    @pytest.mark.need_serialized_dag(False)
+    @conf_vars({("scheduler", "unreferenced_triggers_cleanup_batch_size"): "2"})
+    def test_remove_unreferenced_triggers_drains_in_batches_per_call(self, session):
+        """One scheduler tick removes at most one batch (5 triggers, batch=2 → 3 stay).
+
+        Mirrors the triggerer's cadence (``Trigger.clean_unused`` does one batch per call;
+        later scheduler ticks drain the remainder). Holding locks across unbounded DELETEs
+        would block the triggerer's writers — see #75141.
+        """
+        self.job_runner = SchedulerJobRunner(job=Job())
+
+        for _ in range(5):
+            session.add(
+                Trigger(
+                    classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger",
+                    kwargs={},
+                )
+            )
+        session.commit()
+
+        self.job_runner._remove_unreferenced_triggers(session=session)
+
+        # batch_size=2 ⇒ exactly 2 rows deleted; 3 remain to be drained next tick.
+        assert session.scalar(select(func.count()).select_from(Trigger)) == 3
+
+        # Second tick drains the remaining batch.
+        self.job_runner._remove_unreferenced_triggers(session=session)
+        assert session.scalar(select(func.count()).select_from(Trigger)) == 1
+
+        # Third tick removes the final row.
+        self.job_runner._remove_unreferenced_triggers(session=session)
+        assert session.scalar(select(func.count()).select_from(Trigger)) == 0
+
+    @pytest.mark.need_serialized_dag(False)
+    @pytest.mark.parametrize("batch_size", ["0", "-1"])
+    def test_remove_unreferenced_triggers_rejects_non_positive_batch_size(self, session, batch_size):
+        """A non-positive batch size must raise before any DELETE is issued."""
+        self.job_runner = SchedulerJobRunner(job=Job())
+
+        session.add(
+            Trigger(
+                classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger",
+                kwargs={},
+            )
+        )
+        session.commit()
+
+        with conf_vars({("scheduler", "unreferenced_triggers_cleanup_batch_size"): batch_size}):
+            with pytest.raises(
+                ValueError,
+                match=r"\[scheduler\] unreferenced_triggers_cleanup_batch_size must be at least 1",
+            ):
+                self.job_runner._remove_unreferenced_triggers(session=session)
+
+        assert session.scalar(select(func.count()).select_from(Trigger)) == 1
+
+    @pytest.mark.need_serialized_dag(False)
+    def test_remove_unreferenced_triggers_preserves_referenced_triggers(self, session, dag_maker):
+        """Cleanup must skip triggers still referenced by a deferred TaskInstance.
+
+        Asset-watcher and Callback preservation are covered by the model's own
+        tests in ``tests/unit/models/test_trigger.py``; here we just verify the
+        scheduler delegates the same preservation contract.
+        """
+        self.job_runner = SchedulerJobRunner(job=Job())
+
+        with dag_maker(dag_id="dag", schedule=None, session=session):
+            EmptyOperator(task_id="fake0")
+
+        dr = dag_maker.create_dagrun()
+
+        referenced = Trigger(
+            classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger", kwargs={}
+        )
+        unreferenced = Trigger(
+            classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger", kwargs={}
+        )
+        session.add_all([referenced, unreferenced])
+        session.flush()
+
+        ti = next(iter(dr.task_instances))
+        ti.state = TaskInstanceState.DEFERRED
+        ti.trigger_id = referenced.id
+        session.commit()
+
+        unreferenced_id = unreferenced.id
+        referenced_id = referenced.id
+        self.job_runner._remove_unreferenced_triggers(session=session)
+        session.expire_all()
+
+        remaining_ids = set(session.scalars(select(Trigger.id)).all())
+        assert unreferenced_id not in remaining_ids
+        assert referenced_id in remaining_ids
+
     @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"""