Repository navigation
Conversation
boring-cyborg
Bot
added
area:API
e2e6311 to
c98a233
Compare
ada7de8 to
29b9a8a
Compare
rjgoyln
left a comment
rjgoyln
left a comment
There was a problem hiding this comment.
Overall LGTM. I left a few comments below.
Three non-blocking notes: two on _delete_pending_partitioned_queued_events (the scope of the APDR cleanup and a possible concurrency issue) and one on a helper name.
Just some thoughts from my side — please feel free to resolve them if they’re not applicable.
pierrejeambrun
left a comment
pierrejeambrun
left a comment
There was a problem hiding this comment.
Thanks for picking this up, the semantics are well thought through: the key matches the upstream event, and a partition is only dropped once all its events are gone.
This needs a rebase: #74839 changed DELETE /assets/{asset_id}/queuedEvents to only delete the queued events of Dags the user can edit, and the new partition path should follow that (details inline). I'd also like to see the earlier review comments addressed, especially limiting the cleanup of pending partitions to the ones this call actually touched. If you go with selecting the candidate partitions first, the "no SELECT before DELETE" assertion in test_delete_does_not_read_back_deleted_row_keys will need to change accordingly.
Meant as a regular review, not a change request.
The DELETE queuedEvents endpoints only cleared queued events of non-partitioned assets, so a partitioned event that was wrong or no longer relevant could not be removed and would still trigger the downstream Dag run. They now accept an optional partition_key to delete the queued events of that partition whose Dag run has not been created yet.
The partition_key on the DELETE queued events endpoints now refers to
the partition key of the queued asset event, the same key used by
POST /assets/{asset_id}/materialize, instead of the partition key of
the Dag run the event would create. This lets a single wrong upstream
partition be cleared when a partition mapper rolls several of them up
into one Dag run.
A partitioned asset event row outlives its partition record once the Dag run it created is deleted, since only the partition record cascades. Those rows were matched as queued events, so the endpoint deleted them and returned 204 although nothing was queued. Only events of a partition still waiting for its Dag run are deleted now.
Deleting queued partitioned events swept every empty pending partition in the table, so clearing one asset's key could drop a partition of another Dag that the request never matched. The partitions to clear are now collected before anything is deleted, and both deletes are restricted to them. The asset route now deletes what the caller may edit rather than what it may read, so its test has to tell the two apart.
Two calls clearing different assets of the same partition could each still see the other's uncommitted events under READ COMMITTED, so neither would notice the partition had been emptied and it would stay pending with nothing left to wait for.
29b9a8a to
d37ff6b
Compare
This branch has not been deployed
3 participants
The
DELETEqueued events endpoints only cleared queued events of non-partitioned assets (AssetDagRunQueue). A queued partitioned event (AssetPartitionDagRun+PartitionedAssetKeyLog) that turned out to be wrong or no longer relevant could not be removed, so it would still trigger the downstream Dag run.This adds an optional
partition_keyquery parameter to:DELETE /api/v2/assets/{asset_id}/queuedEventsDELETE /api/v2/dags/{dag_id}/assets/queuedEventsDELETE /api/v2/dags/{dag_id}/assets/{asset_id}/queuedEventsBehaviour when
partition_keyis given:partition_keymatches the partition key of the queued asset event (PartitionedAssetKeyLog.source_partition_key), the same key used byPOST /assets/{asset_id}/materialize. It is not the partition key of the Dag run the event would create, which can differ when the Dag uses a partition mapper. For example, with hourly upstream partitions rolled up into a daily Dag run,partition_key=2026-09-02T05clears only that hour and the daily partition keeps waiting for it to be re-emitted.created_dag_run_id IS NULL) are deleted; partitions that already created a Dag run are kept.AssetDagRunQueue) are not deleted.asset_id,dag_id,before, and readable-Dag filters narrow which events are deleted. A pendingAssetPartitionDagRunis deleted only when all of its contributing events are deleted, so clearing some events of a partition keeps it waiting for them instead of dropping the progress made by the others. An empty pending partition would otherwise sit at the head of the scheduler's FIFO queue forever.Without
partition_keythe endpoints behave exactly as before.closes: #70444
Was generative AI tooling used to co-author this PR?