ferruzzi commented on code in PR #66350:
URL: https://github.com/apache/airflow/pull/66350#discussion_r4039115161
##########
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:
I saw this in the last pass but didn't want to add another round of churn
over a log message but now that you are back in there, I guess we may as well
add it.
Right now when the DELETE removes nothing, `_do_delete` just continues
silently. With the default `skip_archive=False` that leaves the row in two
places: still live in `dag_version`, and also sitting in the archive table that
was just committed. `export-archived` would then emit a row that was never
deleted, and the next cleanup run silently archives it again.
Could you log a warning before the `continue`?
```python
num_rows =
session.scalars(select(func.count()).select_from(limited_query.subquery())).one()
if num_rows == 0: # nothing left to delete
break
```
then in the zero-delete branch:
```python
if deleted == 0:
logger.warning(
"%s rows from %s are still referenced and were not deleted; they
remain in %s",
num_rows,
source_table_name,
target_table_name if not skip_archive else "the archive, which is
being dropped",
)
continue
```
The wording is up to you, as long as a user can tell it happened.
--
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]
