ramitkataria commented on code in PR #66350:
URL: https://github.com/apache/airflow/pull/66350#discussion_r4051362778
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -457,10 +465,37 @@ 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
Review Comment:
The NOT EXISTS is evaluated against the DELETE's own snapshot, right? So if
I understand correctly: if a TI pinning this row is inserted while the DELETE
is already running, the DELETE waits on the row's share lock, then proceeds and
still hits the FK. So this closes the window between the archive commit and the
DELETE starting, but not the DELETE itself.
Would it make sense to also catch `IntegrityError` around the DELETE, roll
back, warn, and `continue`? Roughly the same size as the guard and makes the
newsfragment's "Fixed" accurate. If you'd rather keep just the guard, maybe
reword the newsfragment to say the window is reduced?
##########
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:
What if the SELECT found 50 rows and one got pinned mid-pass, we delete 49,
stay silent, and the pinned row sits in the committed archive looking purged.
Maybe count the archived rows right after the CTAS commit and warn on `deleted
< archived` instead?
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -452,6 +463,27 @@ def test_cleanup_with_dag_id_filtering(self, dag_ids,
exclude_dag_ids, expected_
f"Expected {expected_remaining_dag_ids} to remain, but got
{remaining_dag_ids}"
)
+ @pytest.mark.parametrize(
+ ("dag_ids", "exclude_dag_ids"),
+ [
+ pytest.param(["dag1"], None, id="include"),
+ pytest.param(None, ["dag1"], id="exclude"),
+ ],
+ )
+ def test_cleanup_dag_filtering_on_tables_without_their_own_dag_id(self,
dag_ids, exclude_dag_ids):
Review Comment:
I think we can probably drop this after the rebase? It covers the old scope
on empty tables with no assertions, and `test_no_failure_warnings` in main
already exercises these tables under both flags
##########
airflow-core/newsfragments/71732.bugfix.rst:
##########
@@ -0,0 +1 @@
+Fixed ``airflow db clean`` failing on the ``asset_event``, ``task_reschedule``
and ``deadline`` tables when ``--dag-ids`` or ``--exclude-dag-ids`` was used.
Review Comment:
Should we drop this note now that #73173 has been merged?
--
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]
