This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 67b6015171d [v3-3-test] Reduce memory used when deleting queued asset
events (#71917) (#71937)
67b6015171d is described below
commit 67b6015171db1224e9c1e48fae64491adbe0c3c2
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Aug 22 22:54:00 2026 +0800
[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 49c5d519b0e50e153da9310a91083ccab0f137af)
Co-authored-by: Jyun-An Chen <[email protected]>
---
.../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 c65d4c3dfa5..fd359fd86bc 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
@@ -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(
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 65af5792bef..e1b66fb7a65 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
@@ -2045,6 +2045,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):
@@ -2073,6 +2103,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
