ataulmujeeb-cyber opened a new pull request, #75524: URL: https://github.com/apache/airflow/pull/75524
A partition mapper fans one incoming event out into several downstream keys, and it is free to return them in any order. Each pending run that gets inserted for a batched consumer holds a `(target_dag_id, pending_partition_key)` unique-index entry until its transaction commits, so two producers that take the same keys in opposing orders can wait on each other: 1. Asset A's mapper returns `["k2", "k1"]` for consumer `C`; asset B's mapper returns `["k1", "k2"]`. 2. The producer for A inserts the pending run for `k2`; the producer for B inserts `k1`. 3. A now asks for `k1` and B asks for `k2`. Each waits on an index entry the other holds. Sorting the keys before acquiring gives every producer the same order within one registration, so that particular cycle cannot form. The existing consumer-Dag ordering is untouched, the complete key set is preserved, and the fan-out cap check still runs before acquisition so an over-cap fan-out is still skipped as a whole rather than truncated. Related: #71074, which introduced the pending-run uniqueness this ordering protects. Independent of the error-classifier follow-up and of lifecycle locking; no schema change. This orders keys **within one registration**. A transaction that registers several assets makes several registrations and can still take keys in conflicting orders, so this does not eliminate all deadlocks — it removes the single-registration cycle cheaply. Recovering the remaining cross-registration case is the subject of a separate, larger follow-up. ## Testing The tests drive real asset registration with a mapper that deliberately returns keys in descending order, then assert the complete persisted sequence of pending runs and event links — for one consumer and for two, so the consumer-Dag ordering and the key ordering are both covered. A further case asserts the fan-out cap still skips an over-cap fan-out. ``` pytest airflow-core/tests/unit/assets/test_manager.py -k 'fanout or fan_out or dag_id_order' ``` Run for this submission on the branch head, with `BACKEND` set: 10 passed on PostgreSQL/psycopg3 and 10 passed on MySQL/PyMySQL, no skips. Removing the sort in production makes both new ordering cases fail, so the tests are tied to the change rather than to incidental ordering. <details> <summary>Detailed validation</summary> Recorded during implementation on the equivalent tree, same selection: | Backend / driver | Result | | --- | --- | | PostgreSQL / psycopg3 | 10 passed | | MySQL / PyMySQL | 10 passed | | SQLite | 10 passed | Mutation control: with the production key sort removed, `test_partition_fanout_creates_apdrs_in_key_order` and the multiple-consumer case fail on each of the three backends. ``` pytest airflow-core/tests/unit/assets/test_manager.py::TestAssetManager::test_partition_fanout_creates_apdrs_in_key_order ``` Scope limits worth stating plainly: - This establishes per-registration acquisition order. It is not transaction-wide ordering and not a claim that deadlocks are gone. - The assertions compare full `(dag_id, key)` sequences rather than set membership, because an assertion that only checked membership would pass with the sort removed. - SQLite results do not establish concurrent-writer behaviour. - Early startup attempts in the implementation runs collected no tests because copied provider metadata was stale; that was an environment problem, resolved by regenerating the metadata, and those attempts are excluded from the counts above. </details> --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Codex (GPT-6) and Claude (Opus 5) Generated-by: Codex (GPT-6), with review-driven corrections and publication preparation by Claude (Opus 5), following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) --- * Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged. * For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed. * When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x). * For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). --- Drafted-by: Codex (GPT-6) and Claude (Opus 5) (no human review of this text 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]
