[v3-4-test] Reduce db clean batch probe overhead (#74455)

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)

airflow-core/tests/unit/utils/test_db_cleanup.py

@@ -906,6 +906,26 @@ def test_table_config_dag_id_scope_requires_fk_column(self):
                 # "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"