ferruzzi commented on PR #66350:
URL: https://github.com/apache/airflow/pull/66350#issuecomment-5706047783
Jakub, you've been great about working with my change requests and this is a
bit frustrating, but #73173 merged a couple of hours ago and it fixes the same
three tables this PR does. It came in from the #70923 angle and they didn't
see this existing work. I've left a message over there pointing it out, but
it's already merged.
The DELETE race fix is still a real bug that #73173 doesn't touch, so yours
is still the only fix for it. If you are still up for getting this merged,
here's the last list. Four mechanical bits and two actual code changes.
1. Drop `dag_id_via`, the three `_TableConfig` entries, and
`71732.bugfix.rst`. (( I know, that's the code from arose26 I asked you to
pull into this one in the first place.))
2. Remove `closes: #69030` from your description
3. Change the PR title to reflect the new scope, something like "Fix
`airflow db clean` IntegrityError when a dag_version is referenced by a
task_instance" since that's pretty close to your original commit message and
that's what you'll have left.
4. Rebase on main.
Keep `66350.bugfix.rst`, your `_do_delete` changes, and
`test_dag_id_column_name_matches_schema`. Kaxil's test parametrizes filter
combinations and yours reads the schema directly, so both are worth having.
Also keep the two lines in `_cleanup_table` that pass `skip_if_referenced` and
`referenced_pk_column` through to `_do_delete`.
It should rebase pretty cleanly after that. Then two actual code changes.
On the race test: your 11 Sep version had the sequence right, but it passed
`select(archive_table)`, which doesn't have a NOT EXISTS guard. That's why
`continue` hung. The calling code always runs a guarded query built by
`_build_query`, so a skipped row drops out of the next pass, but an unguarded
SELECT keeps returning that row forever. Switching to `_build_query` fixed the
hang, but now the first pass returns zero rows, the DELETE is never reached,
and the guard isn't actually hit int he test. It needs both: the real
`_build_query` query and the TI inserted mid-pass rather than up front. Try
this:
```python
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.
Reproduces the real race: the dag_version row passes the SELECT
filter and is
archived, and only then does a task_instance referencing it appear.
The
skip_if_referenced guard on the DELETE must skip the row instead of
failing with
IntegrityError, and the loop must still drain because the next
SELECT pass
re-evaluates the same NOT EXISTS guard and excludes it.
"""
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()
raced_old = DagVersion(
dag_id=dag_id,
version_number=1,
bundle_name=bundle_name,
created_at=base_date,
last_updated=base_date,
)
# dag_version is configured keep_last per dag_id, so a lone
version is always the
# keep_last survivor and never eligible. A second, newer
version takes that role
# and leaves raced_old as the deletion candidate.
latest = DagVersion(
dag_id=dag_id,
version_number=2,
bundle_name=bundle_name,
created_at=base_date.add(minutes=1),
last_updated=base_date.add(minutes=1),
)
session.add_all([raced_old, latest])
session.flush()
raced_old_id, latest_id = raced_old.id, latest.id
# Built while nothing references raced_old, so the first SELECT
pass returns it
# and _do_delete archives it.
cfg = config_dict["dag_version"]
query = _build_query(
**cfg.__dict__,
clean_before_timestamp=base_date.add(days=10),
session=session,
)
dag_run = DagRun(dag_id, run_id="race-run",
run_type=DagRunType.MANUAL, start_date=base_date)
ti = create_task_instance(
PythonOperator(task_id="dummy-task", python_callable=print),
run_id=dag_run.run_id,
dag_version_id=raced_old_id,
)
ti.dag_id = dag_id
ti.start_date = base_date
raced = False
def reflect_and_race(tables, session, **kwargs):
"""Insert the referencing TI in the window the race needs.
_do_delete reflects both source and target immediately after
committing
the archive CTAS and immediately before building the DELETE,
so this call
site is the only seam between the two. The MySQL branch
reflects the
target alone earlier in the same pass, hence keying on the
two-table call.
If that call is ever moved or inlined, this test stops
reproducing the
race -- and would pass while testing nothing, so the
``raced`` assertion
below is not optional.
"""
nonlocal raced
if not raced and len(tables) == 2:
raced = True
session.add_all([dag_run, ti])
session.commit()
return reflect_tables(tables, session, **kwargs)
with patch("airflow.utils.db_cleanup.reflect_tables",
side_effect=reflect_and_race):
_do_delete(
query=query,
orm_model=cfg.orm_model,
skip_archive=True,
session=session,
batch_size=None,
skip_if_referenced=cfg.skip_if_referenced,
referenced_pk_column=cfg.referenced_pk_column,
)
remaining =
set(session.scalars(select(DagVersion.id).where(DagVersion.dag_id ==
dag_id)).all())
assert raced, "the TI was never inserted mid-pass; the race was not
reproduced"
assert raced_old_id in remaining, "dag_version referenced by a
task_instance must not be deleted"
assert latest_id in remaining, "the keep_last survivor must not be
deleted"
```
What was happening is that with a single `DagVersion`, the SELECT was
returning zero for two reasons. `dag_version` is `keep_last` per `dag_id`, so
a lone version is always the survivor and is excluded regardless of any TI.
That's why this version creates two. `latest` will take the `keep_last` slot
so `raced_old` comes up for deletion.
I've tested it on postgres and sqlite. It works as written and fails with
the `if skip_if_referenced:` block removed as expected, so I think it should be
good regardless of backend. I don't have a mysql backend handy, but the CI
will test that side.
And the last request. 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. 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.
And honestly, if you've had enough after four months of this, say so. I'll
cherry-pick your work over and get it over the line with your author credit
intact. I'm sorry that this was your first introduction to the project, it's
been a rough one.
--
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]
