uranusjr commented on code in PR #65920:
URL: https://github.com/apache/airflow/pull/65920#discussion_r3257320525
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -226,16 +229,35 @@ def clean_unused(cls, session: Session = NEW_SESSION) ->
None:
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)
+ # Clear task-instance trigger references in primary-key order to avoid
locking the same rows in
+ # a different order than scheduler timeout handling.
+ while True:
+ task_instance_ids = []
+ for attempt in run_with_db_retries():
+ with attempt:
+ candidates = (
+ select(TaskInstance.id)
+ .where(
+ TaskInstance.state != TaskInstanceState.DEFERRED,
+ TaskInstance.trigger_id.is_not(None),
+ )
+ .order_by(TaskInstance.id)
+ .limit(_TRIGGER_ID_CLEANUP_BATCH_SIZE)
+ )
+ task_instance_ids = list(
+ session.scalars(
+ with_row_locks(candidates, of=TaskInstance,
session=session, skip_locked=True)
+ ).all()
)
- .values(trigger_id=None)
- )
+ if task_instance_ids:
+ session.execute(
+ update(TaskInstance)
+ .where(TaskInstance.id.in_(task_instance_ids))
+ .values(trigger_id=None)
+ .execution_options(synchronize_session=False)
+ )
+ if len(task_instance_ids) < _TRIGGER_ID_CLEANUP_BATCH_SIZE:
+ break
Review Comment:
Essntially the same comments as above…
By the way, is it possible to refactor a bit so this logic is not somewhat
duplicated?
--
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]
