[mallesh-bot]

Summary

airflow db clean --skip-archive is documented as a way to avoid the intermediate _airflow_deleted__* archive table. In practice _do_delete always executed the CTAS / INSERT ... SELECT into that archive first and only dropped it in the finally block when skip_archive=True. Every row was written twice, and on deployments with a low statement timeout the DELETE or DROP could time out — the very failure mode #42003 reports (and the motivating cause referenced in the issue).

This PR adds a _delete_directly path that DELETEs each batch's rows in place using an IN-subquery over the same limited_query, and routes skip_archive=True through it. The archive is never created, so there is nothing to drop.

• skip_archive=True → new direct-delete path (one DELETE per batch, one commit per batch — batching preserved from _do_delete).
• skip_archive=False (default) → unchanged archive-then-delete flow.

The direct-delete branches on dialect the same way the archive branch does:

• single-PK tables use pk IN (SELECT pk FROM (limited_query)).
• SQLite composite-PK tables use the same row-value IN form the archive branch does.
• Other dialects (Postgres, MySQL) use EXISTS (SELECT 1 FROM (limited_query) WHERE and_(source.pk == derived.pk)) — mirroring the archive branch's and_(col == target.c[col.name]) join, which is the shape that works on MySQL 5.7 where composite IN (SELECT ...) is rejected.

Test plan

• Two new unit tests in airflow-core/tests/unit/utils/test_db_cleanup.py:
  • test_do_delete_skip_archive_never_creates_archive_table — skip_archive=True routes through _delete_directly and never emits any CTAS/INSERT/DROP.
  • test_do_delete_keep_archive_uses_archive_flow — skip_archive=False still uses the archive path (2 commits, no _delete_directly call).
• Existing TestDBCleanup.test__skip_archive integration test still expects 0 archive tables after skip_archive=True cleanup — the new direct-delete path satisfies that even more strictly than the old create-then-drop path did.
• Four obsolete mock tests were removed: test_do_delete_rolls_back_before_drop_on_failure, test_do_delete_propagates_original_error_when_rollback_fails, test_do_delete_success_does_not_call_rollback, test_do_delete_original_error_survives_archive_drop_failure, test_do_delete_success_propagates_archive_drop_error. Each of these exclusively exercised the if target_table is not None and skip_archive: drop-in-finally branch that airflow db clean --tables dag_version --skip-archive hangs forever (MDL deadlock) when an old dag_version is FK-pinned by task_instance #66177 hardened — that branch no longer exists under this PR (there is no archive to drop). The behaviours those tests protected are no longer reachable, so keeping the tests as-is would mislead about what the code does. The two new tests plus the existing integration test cover the replacement code path.
• Local pytest run blocked on Windows by pykerberos failing to build without GSSAPI headers (documented fallback in AGENTS.md is Breeze, which needs Docker). Ran python -m ast.parse on both modified files and uvx ruff format/ruff check --fix cleanly. CI will run the full suite.

closes: #42003

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.7). I reviewed and understand every change. The design mirrors the approach that maintainer feedback on the earlier attempt in fix: db clean skip-archive creating archive tables #67187 endorsed (that PR was closed for going stale on rebase, not on design), adapted to the current _do_delete shape (which since gained _format_table_name schema qualification and the airflow db clean --tables dag_version --skip-archive hangs forever (MDL deadlock) when an old dag_version is FK-pinned by task_instance #66177 failure-path hardening). Generated per the guidelines.

🤖 Generated with Claude Code

[ferruzzi]

Blocked behind #73568, please do not merge until that one is merged.

---

Just a heads up, this now conflicts with main (PR #66350 is merged and #73568 is following behind it) and the merge resolution has a trap in it. You may as well wait to rebase after #73568 merges, but when you do:

1. Keep the if skip_if_referenced: in _do_delete from main which appends the NOT EXISTS to delete (db_cleanup.py:468 on main). Your PR rewrites the region it lives in, so a "keep my side" resolution would delete that bug fix.

2. Keep the block below it (db_cleanup.py:492): the archived = session.scalars(...) count and the if deleted < archived: warning.

3. This is the step that can go wrong Keep skip_if_referenced= and referenced_pk_column= on the _do_delete(...) call inside _cleanup_table. No test fails if you drop it. Drop those two kwargs and the guard becomes unreachable in production while the tests stay green.

4. Verify with pytest airflow-core/tests/unit/utils/test_db_cleanup.py -k "guards_against_race or pinned_by_task_instance". All three cases must pass. That covers steps 1 and 2 but not step 3, per the above.

5. Please split the chart/ changes into its own PR. This carries chart/values.yaml, chart/templates/_helpers.yaml and two helm test files alongside the core fix, with newsfragments for two unrelated issues (#42003 and #42918). Splitting lets each half be reviewed and backported on its own.

For what it's worth, _delete_directly does not need step 1's re-application: it embeds limited_query as an IN-subquery, so _build_query's NOT EXISTS is re-evaluated at DELETE time. That is load-bearing and currently implicit, so a test pinning it would be worth having.

[Miretpl]

Chart changes here are unrelated to the PR itself.