muditagrawal2007 opened a new issue, #75141:
URL: https://github.com/apache/airflow/issues/75141
### Summary
`SchedulerJobRunner._remove_unreferenced_triggers` issues an **unbounded
single-statement `DELETE` against the `trigger` table every
`parsing_cleanup_interval` (default 60s)**. On any deployment where the trigger
table grows large — which is expected over time, since deferrable operators and
asset/asset-alias watchers each leave a trigger row behind — this holds row
locks for the entire transaction and stalls concurrent writers on the same
table, including the triggerer's own `Trigger.clean_unused()` / `submit_event`
paths. It also forces a full table materialisation in Python via
`synchronize_session="fetch"` before the DELETE runs.
### Why this matters
- **AGENTS.md (Coding Standards) is explicit:** *"Bulk `DELETE`/`UPDATE` in
the scheduler loop or any synchronous interval task (e.g.
`call_regular_interval` callbacks) must be batched with `LIMIT` and committed
between batches — never issue a single unbounded bulk write against a
user-driven table. … Follow the batching pattern in
`airflow-core/src/airflow/utils/db_cleanup.py`."* The current implementation
predates and contradicts that rule. Affected column joins
(`task_instance.trigger_id`, `asset_watcher.trigger_id`, `callback.trigger_id`)
are already indexed.
- **Two cleanup paths, one batched and one not.** #70961 (merge commit
`c3cc5573b8`) moved the **triggerer's** version (`Trigger.clean_unused()`) onto
the existing pattern at `models/trigger.py:_delete_unused_batch`: `select id
LIMIT, with_row_locks(skip_locked=True), delete by id, commit`, driven by
`[triggerer] unreferenced_triggers_cleanup_batch_size`. The scheduler-side
counterpart was missed by that change — only the triggerer's `clean_unused`
callback in `jobs/triggerer_job_runner.py:clean_unused` was rewired. The
scheduler still calls its own `_remove_unreferenced_triggers` from
`jobs/scheduler_job_runner.py:2034`.
### Evidence
`airflow-core/src/airflow/jobs/scheduler_job_runner.py:4206-4219`:
```python
@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")
)
```
Existing tests are green at
`airflow-core/tests/unit/jobs/test_scheduler_job.py:10051` and `:10087`
(covering the unreferenced/referenced-path decision tree) but assume the
unbounded form — they won't fail today because the table is small.
### Why I'm filing this
A previous PR — **#68241** "Batch the scheduler's unreferenced-trigger
cleanup deletes" — implemented the same fix and was closed as a stale draft on
2026-07-02 (author inactive; `potiuk` moved it to draft because static / docs
checks were failing). It would also re-roll the batching logic that already
exists in `Trigger._delete_unused_batch`. **The triggerer side of this work
landed (#70961); the scheduler side did not.**
### Proposed fix
Reuse the helper `Trigger._delete_unused_batch` from the scheduler rather
than re-rolling the LIMIT/with_row_locks/commit-between-batches loop.
Concretely:
1. Add a new config knob `[scheduler]
unreferenced_triggers_cleanup_batch_size` (default `500`, `version_added:
3.3.0` per the existing `[triggerer]` option pattern).
2. Rewrite `_remove_unreferenced_triggers` to read that knob, validate it is
`≥ 1`, and call `Trigger._delete_unused_batch(batch_size, session=session)`.
The triggerer's `clean_unused` will keep the same cadence it has today (one
batch per tick → drain over multiple ticks), so the net behaviour is the same
modulo per-tick lock duration.
3. Extend tests in `tests/unit/jobs/test_scheduler_job.py` with: a
batching-commit-count test (e.g. 5 triggers, `batch_size=2` → 3 commits), a
non-positive-batch-size rejection test mirroring
`tests/unit/models/test_trigger.py:325`, and a sanity test that referenced
triggers survive.
### Areas
- area:scheduler
- area:core
### Reproduction
Hard to reproduce locally on a small fixture, but reproducible on any
long-running install by tailing the scheduler logs for one cleanup tick after
seeding N >> 500 unreferenced trigger rows; you will see the tick stretch to
multiple seconds and any concurrent writer hitting `SELECT … FOR UPDATE` on
`trigger` block.
### Airflow version
main at `7fbcfeea27` (post-#75096).
### Follow-ups
None expected — once the scheduler delegates to
`Trigger._delete_unused_batch`, the cleanup pattern is fully owned by
`Trigger`, and any future tuning (e.g. lock-skip policy, column-level WHERE
tightening) edits one class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
