anmolxlight commented on PR #71074: URL: https://github.com/apache/airflow/pull/71074#issuecomment-5179448143
Pushed a rework that drops the Dag-row lock entirely — no `_lock_target_dag`, no SQLite writer-lock retry loop, no Postgres/MySQL `SELECT ... FOR UPDATE` on `DagModel`. The new approach: a unique constraint on `(target_dag_id, pending_partition_key)`, where `pending_partition_key` mirrors `partition_key` only while `created_dag_run_id IS NULL` (cleared via a `validates` hook the moment a dag run is created). This stays in the design space @potiuk outlined in https://github.com/apache/airflow/issues/58919#issuecomment-3629496888 — no Dag table lock, and no table-wide/advisory lock either. The reason it's a mirrored column instead of the more direct partial/filtered unique index (`WHERE created_dag_run_id IS NULL`) is portability: MySQL supports neither partial nor filtered indexes, so the mirrored-null-column trick is what's actually usable across sqlite/Postgres/MySQL — a plain unique index already treats NULL as distinct from every other value on all three. Concurrency is now optimistic instead of pessimistic: `_get_or_create_apdr` selects the latest pending APDR, and if none exists, inserts inside a `SAVEPOINT` (`session.begin_nested()`). If two threads race, the loser's INSERT hits the unique constraint and raises `IntegrityError`, which is caught to re-select and work on the winner's row instead of raising — consistent with the model's existing "always work on the latest matching APDR record" contract. Migration `0128_3_4_0_add_pending_partition_key_to_apdr.py` adds the column, backfills it, collapses any pre-existing duplicate pending rows down to the latest one per key (mirroring the scheduler's own stale-APDR cleanup), then creates the constraint. Test evidence: `test_get_or_create_apdr_is_idempotent_under_concurrent_calls` now asserts convergence on exactly one row without assuming which of the two code paths (initial SELECT hit vs. losing the INSERT race) each thread takes, since that's a timing detail, not the invariant. `test_register_asset_change_from_different_producer_assets_does_not_duplicate_apdr` (the apache/airflow#71070 regression test) now exercises the real path with no lock spy. Added `test_pending_partition_key_unique_constraint_blocks_duplicate_pending_rows` (DB rejects a second pending row directly) and `test_created_dag_run_id_assignment_clears_pending_partition_key` (validator coverage). Full `airflow-core/tests/unit/assets/test_manager.py` suite: 56 passed. --- Drafted-by: Claude Code (Sonnet 5) (no human review before posting) -- 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]
