ferruzzi commented on code in PR #66350:
URL: https://github.com/apache/airflow/pull/66350#discussion_r4074908658
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -392,10 +418,32 @@ def _do_delete(
delete = source_table.delete().where(
and_(*[col == target_table.c[col.name] for col in
source_table.primary_key.columns])
)
+ # Re-apply skip_if_referenced on the DELETE to guard against a
race where a new
+ # referencing row is created after the archive INSERT committed
but before the DELETE
+ # runs. Without this the DELETE would violate the ON DELETE
RESTRICT FK and fail.
+ if skip_if_referenced:
+ pk_col = source_table.c[referenced_pk_column]
+ for referencing_table_name, fk_column in skip_if_referenced:
+ referencing = table(referencing_table_name,
column(fk_column))
+ delete = delete.where(
+ ~select(literal(1))
+ .select_from(referencing)
+ .where(referencing.c[fk_column] == pk_col)
+ .correlate(source_table)
+ .exists()
+ )
logger.debug("delete statement:\n%s", delete.compile())
- session.execute(delete)
+ deleted = cast("CursorResult", session.execute(delete)).rowcount
session.commit()
+ # A guarded DELETE (skip_if_referenced) may delete fewer rows than
the SELECT
+ # found. That is fine: the SELECT already includes the same NOT
EXISTS guard, so
+ # the skipped row is excluded on the next pass too, and the loop
drains naturally.
+ # With --batch-size set, continuing here lets subsequent batches
clean rows that
+ # were not affected by the race.
+ if deleted == 0:
Review Comment:
Actually, I'll put that in another followup, I already have one which is
gated behind #73417 (which this one is blocking) and that is a better place for
it anyway. I'm merging this. Thanks for all your work here.
--
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]
