Repository navigation
Avoid unbounded IN lists for queued asset events - #75943
Conversation
Each asset_dag_run_queue row now references a single asset event, so a Dag can have any number of queued rows. The scheduler sent them in a literal IN, and the statement would exceed database limits on bind parameters or expression depth if there are too many rows. Those queries now use subqueries bounded by the largest event id so the query's bind parameter count is constant. As a result, a run may consume queued events whose rows were skipped by SKIP LOCKED if their ids are within the bound. Since only one scheduler can hold an event row (the Dag row itself is locked), this should only affect rows being dequeued concurrently elsewhere (e.g. with the API). Even in this scenario, the end result should still be correct, and the scheduler's existing retry mechanism should recover potential deadlock issues.
6ad02ca to
d69dccb
Compare
Backport successfully created: v3-4-testNote: As of Merging PRs targeted for Airflow 3.X In matter of doubt please ask in #release-management Slack channel.
|
| """ | ||
| stmt = delete(AssetDagRunQueue).where( | ||
| AssetDagRunQueue.target_dag_id == dag_id, | ||
| AssetDagRunQueue.asset_event_id <= max_event_id, |
There was a problem hiding this comment.
This range now also matches rows SKIP LOCKED skipped, and a DELETE waits on a locked row instead of skipping it. If one of the queuedEvents DELETE endpoints has locked row A and is waiting on a row this scheduler holds, this statement reaching A closes the cycle. A Postgres repro with the timing forced by a sleep gets DeadlockDetected at this head, while on the base commit the API call just waits. The description says the scheduler's retry recovers, but Postgres aborts whichever backend runs the deadlock check first, which can be the API request, and that request gets a 500 with no retry. Could the DELETE be limited to rows this transaction locked, e.g. asset_event_id IN (SELECT ... WHERE target_dag_id = :d AND asset_event_id <= :max FOR UPDATE SKIP LOCKED)? The parameter count stays constant, though MySQL needs a derived table for the self-reference.
| Select unconsumed events referenced by a Dag's locked ADRQ rows. | ||
|
|
||
| *max_event_id* is the largest event ID among the rows locked by this | ||
| scheduler, so events queued after the lock was taken are left for the |
There was a problem hiding this comment.
I don't think the bound guarantees this. With catchup=True the two OR'd arms below have no id bound, so they still pick up events queued after the lock. And event ids aren't assigned in commit order, so on Postgres a lower id can commit after the lock and still fall inside the bound. Both behave the same as the old code, so it's only the docstring (and the max_event_id param doc on _delete_consumed_asset_records) that overpromises.
| pytest.param(1, {0, 1}, set(), id="skipped-row-older"), | ||
| ], | ||
| ) | ||
| def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows( |
There was a problem hiding this comment.
In the skipped-row-older case the skipped row is consumed and dequeued too, so "deletes only selected" no longer describes what this asserts. Maybe rename it to something like ..._consumes_skipped_rows_within_bound?
| def test_create_dagruns_asset_triggered_many_queued_events_no_literal_in( | ||
| session: Session, dag_maker, batch_asset_events: bool | ||
| ): | ||
| """Queued events must not be sent to the database as a literal list (#75865).""" |
There was a problem hiding this comment.
Could we drop the (#75865) here and in the docstring on line 13134? The test name and docstring already describe the behaviour, and the PR links the issue.
Fix issue mentioned in #75865 (comment)
Each
asset_dag_run_queuerow now references a single asset event, so a Dag can have any number of queued rows. The scheduler sent them in a literal IN, and the statement would exceed database limits on bind parameters or expression depth if there are too many rows.Those queries now use subqueries bounded by the largest event id so the query’s bind parameter count is constant.
As a result, a run may consume queued events whose rows were skipped by SKIP LOCKED if their ids are within the bound. Since only one scheduler can hold an event row (the Dag row itself is locked), this should only affect rows being dequeued concurrently elsewhere (e.g. with the API). Even in this scenario, the end result should still be correct, and the scheduler's existing retry mechanism should recover potential deadlock issues.