This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 4fbbcddf022 Reduce db clean batch probe overhead (#74455)
4fbbcddf022 is described below
commit 4fbbcddf02245c38dd7ea369fa4b16fa7d9447f4
Author: safaehar <[email protected]>
AuthorDate: Sun Oct 18 08:37:59 2026 +0200
Reduce db clean batch probe overhead (#74455)
Large cleanups repeat the batch probe many times, and counting every row up
to the batch size adds unnecessary database work.
---
airflow-core/src/airflow/utils/db_cleanup.py | 4 +---
airflow-core/tests/unit/utils/test_db_cleanup.py | 20 ++++++++++++++++++++
2 files changed, 21 insertions(+), 3 deletions(-)
diff --git a/airflow-core/src/airflow/utils/db_cleanup.py
b/airflow-core/src/airflow/utils/db_cleanup.py
index 02ba0975d5c..23ea7e41bc1 100644
--- a/airflow-core/src/airflow/utils/db_cleanup.py
+++ b/airflow-core/src/airflow/utils/db_cleanup.py
@@ -496,9 +496,7 @@ def _do_delete(
while True:
limited_query = query.limit(batch_size) if batch_size else query
- if (
-
session.scalars(select(func.count()).select_from(limited_query.subquery())).one()
== 0
- ): # nothing left to delete
+ if not session.scalars(select(limited_query.exists())).one(): #
nothing left to delete
break
batch_no = next(batch_counter)
diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py
b/airflow-core/tests/unit/utils/test_db_cleanup.py
index 38daeda56bf..045a1587977 100644
--- a/airflow-core/tests/unit/utils/test_db_cleanup.py
+++ b/airflow-core/tests/unit/utils/test_db_cleanup.py
@@ -906,6 +906,26 @@ class TestDBCleanup:
# "dagrun_id" intentionally omitted from extra_columns
)
+ def test_do_delete_checks_batch_existence_without_counting_rows(self):
+ session = MagicMock(spec=Session)
+ session.get_bind.return_value.dialect.name = "sqlite"
+ session.scalars.return_value.one.return_value = False
+ _, source_table, _, query = _build_do_delete_test_objects()
+
+ _do_delete(
+ query=query,
+ orm_model=source_table,
+ skip_archive=False,
+ session=session,
+ batch_size=100,
+ )
+
+ existence_query = session.scalars.call_args.args[0]
+ compiled_query = str(existence_query.compile()).lower()
+ assert "exists" in compiled_query
+ assert "count(" not in compiled_query
+ session.execute.assert_not_called()
+
def test_do_delete_rolls_back_before_drop_on_failure(self):
session = MagicMock(spec=Session)
session.get_bind.return_value.dialect.name = "mysql"
