shivaam commented on code in PR #70961:
URL: https://github.com/apache/airflow/pull/70961#discussion_r4134179281
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -231,42 +231,54 @@ def fetch_trigger_ids_with_non_task_associations(cls, *,
session: Session = NEW_
return set(session.scalars(query))
@classmethod
- @provide_session
- def clean_unused(cls, *, session: Session = NEW_SESSION) -> None:
+ def clean_unused(cls) -> None:
"""
- Delete all triggers that have no tasks dependent on them and are not
associated to an asset.
+ Delete triggers that have no dependent tasks, assets, or callbacks in
bounded transactions.
Triggers have a one-to-many relationship to task instances, so we need
to clean those up first.
Afterward we can drop the triggers not referenced by anyone.
"""
- # Update all task instances with trigger IDs that are not DEFERRED to
remove them
- for attempt in run_with_db_retries():
- with attempt:
- session.execute(
- update(TaskInstance)
- .where(
- TaskInstance.state != TaskInstanceState.DEFERRED,
TaskInstance.trigger_id.is_not(None)
- )
- .values(trigger_id=None)
- )
-
- # 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(),
- ~cls.task_instance.has(),
- )
+ batch_size = conf.getint("triggerer",
"unreferenced_triggers_cleanup_batch_size", fallback=500)
+ if batch_size <= 0:
+ raise ValueError("[triggerer]
unreferenced_triggers_cleanup_batch_size must be at least 1")
+
+ clear_task_instance_references = True
+ while True:
Review Comment:
Thanks, updated this to delete at most one batch per `clean_unused()` call,
allowing runner servicing and heartbeat before the next batch. A regression
test verifies that five rows with a batch size of two drain across three calls
(`2`, `2`, `1`).
One trade-off is bounded cleanup throughput. For example, if other
supervisor work takes 30 seconds and deleting one 500-row batch takes 10
seconds, cleanup capacity is about 500 rows every 40 seconds, or 12.5 rows per
second. If unused triggers accumulate faster than that, the backlog grows and
the configurable batch size may need tuning.
I’m keeping the `callback.trigger_id` index as a separate follow-up, as
discussed.
--
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]
