ataulmujeeb-cyber opened a new pull request, #75523: URL: https://github.com/apache/airflow/pull/75523
Creating a pending partitioned asset run treats every `IntegrityError` as "another producer inserted this row first". Only one constraint actually means that, so unrelated database errors are swallowed into the race-recovery path. The clearest failure is when the error comes from the caller's own pending rows rather than from the new row. `Session.begin_nested()` flushes pending ORM state before opening its savepoint, so a caller that has queued an invalid row triggers the flush here: 1. A caller queues a row that violates `NOT NULL` and calls into asset-change registration. 2. `begin_nested()` flushes it; the database rejects it and the transaction is now in a failed state. 3. Recovery assumes it lost an insert race and issues a lookup for the "winner". 4. That lookup runs on the failed transaction, so SQLAlchemy raises `PendingRollbackError`. The original error is replaced by one that points at the wrong place, and an error that cannot be fixed by retrying is retried first. This restricts recovery to the pending-marker unique constraint, read from driver diagnostics rather than the rendered message: SQLSTATE `23505` plus the constraint name on PostgreSQL, `sqlite_errorcode` 2067 with the column list on SQLite, and errno 1062 with the trailing `for key` name on MySQL. Anything else propagates unchanged. Genuine marker conflicts keep the existing two-attempt recovery and the existing exhaustion behaviour of re-raising the driver's `IntegrityError`. Reading diagnostics rather than text matters because the rendered `IntegrityError` includes the statement and its bound parameters, so an ordinary row whose value happens to contain the constraint name, or a MySQL value containing its own `for key '...'` fragment, is enough to make an unrelated failure look like the marker conflict. Related: #71074, which added the marker. This is an independent follow-up and changes no schema; it does not touch fan-out ordering or lifecycle locking. ## Testing The tests provoke real constraint violations through the database rather than constructing exceptions, so the same assertions run on every backend. They cover genuine marker conflicts, unrelated primary-key, `NOT NULL` and `UNIQUE` failures, adversarial values containing the constraint name or a misleading `for key` fragment, helper-level propagation and recovery, and preservation of a caller-flush failure. ``` pytest airflow-core/tests/unit/assets/test_marker_conflict_classifier.py ``` Run for this submission on the branch head, with `BACKEND` set: 11 passed on PostgreSQL/psycopg3 and 11 passed on MySQL/PyMySQL, no skips. Removing the guard — restoring the bare `except IntegrityError` — makes the two negative helper tests fail: the caller-flush case performs three lookups instead of one and surfaces `PendingRollbackError` in place of the original error. <details> <summary>Detailed validation</summary> Earlier validation of the same change, recorded during implementation on the equivalent tree: | Backend / driver | Selection | Result | | --- | --- | --- | | PostgreSQL / psycopg3 | focused classifier + helper selection | 22 passed | | PostgreSQL / psycopg2 | `test_marker_conflict_classifier.py` | 11 passed, plus 8 Core-insert fixture controls | | MySQL / mysqlclient | `test_marker_conflict_classifier.py` | 11 passed | | MySQL / PyMySQL | `test_marker_conflict_classifier.py` | 11 passed | | SQLite | focused selection | 21 passed, 1 pre-existing concurrency test skipped | Mutation control, run on PostgreSQL/psycopg3, MySQL/mysqlclient and SQLite: with the bare catch restored, `test_helper_propagates_an_unrelated_integrity_error` and the caller-flush control both fail on each backend. Notes on what this evidence does and does not establish: - `test_helper_recovers_from_a_real_conflict_end_to_end` forces the first lookup to miss so the insert hits a genuine unique violation. It is a staged single-transaction control, not a two-transaction race. - Obscuring the original error is a diagnostic problem. These tests do not demonstrate data loss. - An initial PyMySQL run reported a fixture error; after correcting the fixture the module passed. That startup failure is excluded from the counts above. - SQLite results do not establish concurrent-writer behaviour. </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]
