ferruzzi commented on code in PR #66350:
URL: https://github.com/apache/airflow/pull/66350#discussion_r4019995582
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -392,10 +417,31 @@ 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 = session.execute(delete).rowcount
session.commit()
+ # A guarded DELETE (skip_if_referenced) may affect fewer rows than
the SELECT
+ # saw. If it affects zero, the next SELECT would return the same
rows and we
+ # would archive them forever, so break out. This also caps the
loop in any
+ # edge case where the DELETE unexpectedly removes nothing.
+ if deleted == 0:
+ break
Review Comment:
Just to clarify, "that was my suggestion" was phrased poorly. I meant to
say that `break` was my suggestion, you did exactly what I asked, that was my
mistake. :+1:
--
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]
