[v3-3-test] Reduce memory used when deleting queued asset events (#71917) (#71937)

delete_asset_queued_events and delete_dag_asset_queued_event forced
SQLAlchemy's "fetch" synchronize_session strategy on their
AssetDagRunQueue deletes. That strategy reads the primary key of every
deleted row back from the database to update the ORM session's
identity map, but neither endpoint loads any AssetDagRunQueue objects
into the session beforehand, so the read-back keys are matched against
an empty map and discarded.

The sibling delete_dag_asset_queued_events endpoint already used the
default "auto" strategy; the other two now match it.
(cherry picked from commit 49c5d51)

Co-authored-by: Jyun-An Chen <jun930436@gmail.com>

Author: github-actions[bot]

airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

@@ -667,7 +667,7 @@ def delete_asset_queued_events(
     where_clause = _generate_queued_event_where_clause(
         asset_id=asset_id, before=before, permitted_dag_ids=readable_dags_filter.value
     )
-    delete_stmt = delete(AssetDagRunQueue).where(*where_clause).execution_options(synchronize_session="fetch")
+    delete_stmt = delete(AssetDagRunQueue).where(*where_clause)
     result = cast("CursorResult", session.execute(delete_stmt))
     if result.rowcount == 0:
         raise HTTPException(
@@ -734,9 +734,7 @@ def delete_dag_asset_queued_event(
     where_clause = _generate_queued_event_where_clause(
         dag_id=dag_id, before=before, asset_id=asset_id, permitted_dag_ids=readable_dags_filter.value
     )
-    delete_statement = (
-        delete(AssetDagRunQueue).where(*where_clause).execution_options(synchronize_session="fetch")
-    )
+    delete_statement = delete(AssetDagRunQueue).where(*where_clause)
     result = cast("CursorResult", session.execute(delete_statement))
     if result.rowcount == 0:
         raise HTTPException(

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py

@@ -2045,6 +2045,36 @@ def test_should_respond_404(self, test_client):
         assert response.status_code == 404
         assert response.json()["detail"] == "Queue event with asset_id: `1` was not found"
 
+    def test_delete_does_not_read_back_deleted_row_keys(self, test_client, session, create_dummy_dag):
+        from sqlalchemy import event
+
+        import airflow.settings
+
+        dag, _ = create_dummy_dag()
+        dag_id = dag.dag_id
+        (asset,) = self.create_assets(session=session, num=1)
+        self._create_asset_dag_run_queues(dag_id, asset.id, session)
+
+        executed_statements: list[str] = []
+
+        def capture(_conn, _cursor, statement, _parameters, _context, _executemany):
+            executed_statements.append(" ".join(statement.split()).upper())
+
+        event.listen(airflow.settings.engine, "before_cursor_execute", capture)
+        try:
+            response = test_client.delete(f"/assets/{asset.id}/queuedEvents")
+        finally:
+            event.remove(airflow.settings.engine, "before_cursor_execute", capture)
+
+        assert response.status_code == 204
+        deletes = [s for s in executed_statements if s.startswith("DELETE")]
+        assert deletes, "Expected the endpoint to issue a DELETE statement"
+        assert [s for s in deletes if "RETURNING" in s] == [], "DELETE must not read back deleted keys"
+        after_first_delete = executed_statements[executed_statements.index(deletes[0]) :]
+        assert [s for s in after_first_delete if s.startswith("SELECT")] == [], (
+            "No SELECT may precede a DELETE to collect the keys it is about to remove"
+        )
+
 
 class TestDeleteDagAssetQueuedEvent(TestQueuedEventEndpoint):
     def test_delete_should_respond_204(self, test_client, session, create_dummy_dag):
@@ -2073,6 +2103,36 @@ def test_should_respond_403(self, unauthorized_test_client):
         response = unauthorized_test_client.delete("/dags/random/assets/random/queuedEvents")
         assert response.status_code == 403
 
+    def test_delete_does_not_read_back_deleted_row_keys(self, test_client, session, create_dummy_dag):
+        from sqlalchemy import event
+
+        import airflow.settings
+
+        dag, _ = create_dummy_dag()
+        dag_id = dag.dag_id
+        (asset,) = self.create_assets(session=session, num=1)
+        self._create_asset_dag_run_queues(dag_id, asset.id, session)
+
+        executed_statements: list[str] = []
+
+        def capture(_conn, _cursor, statement, _parameters, _context, _executemany):
+            executed_statements.append(" ".join(statement.split()).upper())
+
+        event.listen(airflow.settings.engine, "before_cursor_execute", capture)
+        try:
+            response = test_client.delete(f"/dags/{dag_id}/assets/{asset.id}/queuedEvents")
+        finally:
+            event.remove(airflow.settings.engine, "before_cursor_execute", capture)
+
+        assert response.status_code == 204
+        deletes = [s for s in executed_statements if s.startswith("DELETE")]
+        assert deletes, "Expected the endpoint to issue a DELETE statement"
+        assert [s for s in deletes if "RETURNING" in s] == [], "DELETE must not read back deleted keys"
+        after_first_delete = executed_statements[executed_statements.index(deletes[0]) :]
+        assert [s for s in after_first_delete if s.startswith("SELECT")] == [], (
+            "No SELECT may precede a DELETE to collect the keys it is about to remove"
+        )
+
     def test_should_respond_404(self, test_client):
         dag_id = "not_exists"
         asset_id = 1