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 b61c9c63298 Reduce memory used when deleting a Dag with a large
history (#71185)
b61c9c63298 is described below
commit b61c9c63298a6f171bd04cd4090e9d1b4387d896
Author: Jyun-An Chen <[email protected]>
AuthorDate: Thu Aug 20 22:56:18 2026 +0800
Reduce memory used when deleting a Dag with a large history (#71185)
delete_dag forced SQLAlchemy's "fetch" synchronization strategy on every
bulk
delete it issues. That strategy reads the primary key of every deleted row
back
from the database so it can mark matching in-memory objects as deleted, but
the
session holds nothing beyond the Dag's own DagModel row — the keys were
matched
against an effectively empty identity map and discarded, once per table
with a
dag_id column.
The cost scaled with the Dag's history rather than with the number of
objects
actually needing synchronization: roughly 211 bytes of transient Python
heap per
deleted row on PostgreSQL, or about 1 GiB in the API server for a Dag with
five
million task instances.
The default strategy evaluates the criteria in Python against the objects
already
loaded, so synchronization still happens without a round-trip sized by the
row
count.
---
airflow-core/src/airflow/api/common/delete_dag.py | 8 +--
.../tests/unit/api/common/test_delete_dag.py | 78 ++++++++++++++++++++++
2 files changed, 80 insertions(+), 6 deletions(-)
diff --git a/airflow-core/src/airflow/api/common/delete_dag.py
b/airflow-core/src/airflow/api/common/delete_dag.py
index 0e9979762d8..e32e8bb90f3 100644
--- a/airflow-core/src/airflow/api/common/delete_dag.py
+++ b/airflow-core/src/airflow/api/common/delete_dag.py
@@ -74,21 +74,17 @@ def delete_dag(dag_id: str, keep_records_in_log: bool =
True, *, session: Sessio
count: int = 0
for model in models_for_deletion:
if hasattr(model, "dag_id") and (not keep_records_in_log or
model.__name__ != "Log"):
- result: Result = session.execute(
- delete(model).where(model.dag_id ==
dag_id).execution_options(synchronize_session="fetch")
- )
+ result: Result = session.execute(delete(model).where(model.dag_id
== dag_id))
cursor_result = cast("CursorResult", result)
count += cursor_result.rowcount
# Delete entries in Import Errors table for a deleted Dag
# This handles the case when the dag_id is changed in the file
session.execute(
- delete(ParseImportError)
- .where(
+ delete(ParseImportError).where(
ParseImportError.filename == dag.relative_fileloc,
ParseImportError.bundle_name == dag.bundle_name,
)
- .execution_options(synchronize_session="fetch")
)
return count
diff --git a/airflow-core/tests/unit/api/common/test_delete_dag.py
b/airflow-core/tests/unit/api/common/test_delete_dag.py
new file mode 100644
index 00000000000..eea327bdfcd
--- /dev/null
+++ b/airflow-core/tests/unit/api/common/test_delete_dag.py
@@ -0,0 +1,78 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+from sqlalchemy import func, select
+
+from airflow.api.common.delete_dag import delete_dag
+from airflow.models import DagModel
+from airflow.providers.standard.operators.empty import EmptyOperator
+
+if TYPE_CHECKING:
+ from airflow.serialization.definitions.dag import SerializedDAG
+
+ from tests_common.pytest_plugin import DagMaker
+
+pytestmark = [pytest.mark.db_test, pytest.mark.need_serialized_dag]
+
+DAG_ID = "dag_to_delete"
+
+
+def test_delete_dag_does_not_read_back_deleted_row_keys(dag_maker:
DagMaker[SerializedDAG], session):
+ """
+ delete_dag must not ask the database for the keys of the rows it deletes.
+
+ Forcing SQLAlchemy's "fetch" synchronization strategy reads every deleted
primary key
+ back, which costs memory proportional to the Dag's history. Backends with
RETURNING
+ stream those keys back on the DELETE itself, and those without it (MySQL)
run a
+ second full SELECT beforehand, so both shapes are asserted against here.
+ """
+ from sqlalchemy import event
+
+ import airflow.settings
+
+ with dag_maker(DAG_ID, session=session):
+ EmptyOperator(task_id="task")
+ dag_maker.create_dagrun()
+ session.commit()
+
+ 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:
+ delete_dag(DAG_ID, keep_records_in_log=False, session=session)
+ session.commit()
+ finally:
+ event.remove(airflow.settings.engine, "before_cursor_execute", capture)
+
+ deletes = [s for s in executed_statements if s.startswith("DELETE")]
+ assert deletes, "Expected delete_dag to issue DELETE statements"
+ assert [s for s in deletes if "RETURNING" in s] == [], "DELETEs 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"
+ )
+
+ assert
session.scalar(select(func.count()).select_from(DagModel).where(DagModel.dag_id
== DAG_ID)) == 0
