ferruzzi commented on code in PR #66350:
URL: https://github.com/apache/airflow/pull/66350#discussion_r3984091713
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -551,6 +655,78 @@ def
test_dag_version_cleanup_skips_versions_pinned_by_task_instance(self):
assert latest_id in remaining # kept by keep_last
assert orphan_id not in remaining # old and unreferenced -> pruned
+ def test_do_delete_skip_if_referenced_guards_against_race(self):
+ """_do_delete must not issue a DELETE that violates an ON DELETE
RESTRICT FK.
+
+ Simulates a race where a dag_version row passes the SELECT filter (no
TI
+ references it at archive-creation time) but a TI referencing it is
inserted
+ before the DELETE runs. The skip_if_referenced guard on the DELETE
itself
+ must leave the row in place instead of failing with IntegrityError.
+ """
+ from airflow.utils.db import reflect_tables
+
+ base_date = pendulum.DateTime(2020, 1, 1,
tzinfo=pendulum.timezone("UTC"))
+ bundle_name = f"race-test-{uuid4()}"
+ dag_id = f"race_dag_{uuid4()}"
+
+ with create_session() as session:
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+ session.add(DagModel(dag_id=dag_id, bundle_name=bundle_name))
+ session.flush()
+
+ dv = DagVersion(
+ dag_id=dag_id,
+ version_number=1,
+ bundle_name=bundle_name,
+ created_at=base_date,
+ last_updated=base_date,
+ )
+ session.add(dv)
+ session.flush()
+ dv_id = dv.id
+
+ # Manually create an archive table containing this dag_version row,
+ # simulating the CTAS step that ran before the TI was inserted.
+ archive_name = f"{ARCHIVE_TABLE_PREFIX}dag_version__race_test"
+ session.execute(
+ text(f"CREATE TABLE {archive_name} AS SELECT * FROM
dag_version WHERE id = '{dv_id}'")
Review Comment:
I didn't notice this until looking into your failing CI to see if they were
real failures or flake. I'm still brushing up on database stuff, so I may be
mistaken, but this test can't pass on Postgres, and on the other two backends
it passes without testing anything.
`dv_id` is a `uuid.UUID`, so this f-string is the standard 36 characters
with hyphens. `DagVersion.id` is `sa.Uuid()`, which uses a native column type
only on Postgres; elsewhere SQLAlchemy stores `value.hex`, which is 32 chars
with the hyphens stripped.
So on Postgres the row does get a match, but `_do_delete` deletes from
`dag_version`, not from `archive_table`. The `while True` loop at
`db_cleanup.py:359` can't reach its `count == 0` condition and just churns
until it hits timeout. That is the five red tests; it's just this test running
the various Postgress combinations.
The interesting side is that ((happy to be wrong here, if you know better))
on MySQL and Sqlite this `CREATE TABLE ... AS SELECT` can't match any rows
because of the hyphenated/unhyphenated mismatch. The `archive` table is
created empty, `_do_delete` returns 0 without anything having been done, and
the `assert` passes.
In the end, the `skip_if_referenced-on-DELETE` guard doesn't actually get
tested. Binding the UUID as a parameter fixes the false passes, but I'm
reasonably sure they'll also just start timing out. Adding `if
session.execute(delete).rowcount == 0: break` after the DELETE should let the
loop finish and would remove a real unbounded-loop hazard. The mock `session`s
on lines 817 and 880 would then need a `rowcount`, and maybe a few other small
tweaks.
--
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]
