This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 49c5d519b0e Reduce memory used when deleting queued asset events
(#71917)
49c5d519b0e is described below
commit 49c5d519b0e50e153da9310a91083ccab0f137af
Author: Jyun-An Chen <[email protected]>
AuthorDate: Fri Aug 21 18:36:37 2026 +0800
Reduce memory used when deleting queued asset events (#71917)
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.
---
.../api_fastapi/core_api/routes/public/assets.py | 6 +--
.../core_api/routes/public/test_assets.py | 60 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 4 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
index 14c40480231..0a30d719d1a 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -711,7 +711,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(
@@ -778,9 +778,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(
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index 1ae3690e19d..d2a60752281 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
@@ -2569,6 +2569,36 @@ class
TestDeleteAssetQueuedEvents(TestQueuedEventEndpoint):
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):
@@ -2597,6 +2627,36 @@ class
TestDeleteDagAssetQueuedEvent(TestQueuedEventEndpoint):
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
