From 6f9dc257c40fcd31821cafb107b14aad3dfef820 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Fri, 3 Jul 2026 01:44:17 +0000 Subject: [PATCH 1/9] Fix airflow db clean hang on MySQL when delete fails (#66177) --- airflow-core/newsfragments/66177.bugfix.rst | 1 + airflow-core/src/airflow/utils/db_cleanup.py | 8 +- .../tests/unit/utils/test_db_cleanup.py | 213 +++++++++++++++++- 3 files changed, 215 insertions(+), 7 deletions(-) create mode 100644 airflow-core/newsfragments/66177.bugfix.rst diff --git a/airflow-core/newsfragments/66177.bugfix.rst b/airflow-core/newsfragments/66177.bugfix.rst new file mode 100644 index 0000000000000..dcf9e601c70d6 --- /dev/null +++ b/airflow-core/newsfragments/66177.bugfix.rst @@ -0,0 +1 @@ +Fix ``airflow db clean --skip-archive`` hanging indefinitely on MySQL when a delete fails due to a foreign-key constraint. The session now rolls back before dropping the temporary archive table, so the command fails fast with the database error instead of blocking on metadata locks. diff --git a/airflow-core/src/airflow/utils/db_cleanup.py b/airflow-core/src/airflow/utils/db_cleanup.py index 8197d56c42deb..25901a8a50a4e 100644 --- a/airflow-core/src/airflow/utils/db_cleanup.py +++ b/airflow-core/src/airflow/utils/db_cleanup.py @@ -323,12 +323,12 @@ def _do_delete( session.execute(delete) session.commit() - except BaseException as e: - raise e + except BaseException: + session.rollback() + raise finally: if target_table is not None and skip_archive: - bind = session.get_bind() - target_table.drop(bind=bind) + target_table.drop(bind=session.connection()) session.commit() print("Finished Performing Delete") diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index 4d66a96b1a358..050d034c2dcc8 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -17,18 +17,21 @@ # under the License. from __future__ import annotations +import threading +import time from contextlib import suppress from importlib import import_module from io import StringIO from pathlib import Path -from unittest.mock import MagicMock, mock_open, patch +from unittest.mock import MagicMock, call, mock_open, patch from uuid import uuid4 import pendulum import pytest -from sqlalchemy import func, inspect, select, text -from sqlalchemy.exc import OperationalError, SQLAlchemyError +from sqlalchemy import Column, Integer, MetaData, Table, func, inspect, literal, select, text +from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError from sqlalchemy.ext.declarative import DeclarativeMeta +from sqlalchemy.orm import Session from airflow import DAG from airflow._shared.timezones import timezone @@ -46,6 +49,7 @@ _build_query, _cleanup_table, _confirm_drop_archives, + _do_delete, _dump_table_to_file, _get_archived_table_names, _TableConfig, @@ -558,6 +562,79 @@ def test_table_config_skip_if_referenced_requires_pk_column(self): # "id" intentionally omitted from extra_columns ) + def test_do_delete_rolls_back_before_drop_on_failure(self): + session = MagicMock(spec=Session) + session.get_bind.return_value.dialect.name = "mysql" + session.connection.return_value = object() + session.scalars.return_value.one.side_effect = [1, 0] + tracker = MagicMock() + tracker.attach_mock(session, "session") + + metadata, source_table, target_table, query = _build_do_delete_test_objects() + delete_failure = IntegrityError("DELETE FROM dag_version", {}, Exception("fk violation")) + session.execute.side_effect = [None, None, delete_failure] + + with ( + patch("airflow.utils.db_cleanup.reflect_tables", return_value=metadata), + patch("airflow.utils.db_cleanup.timezone.utcnow", return_value=_delete_test_timestamp()), + patch.object(target_table, "drop") as drop_mock, + ): + tracker.attach_mock(drop_mock, "drop") + with pytest.raises(IntegrityError) as exc_info: + _do_delete( + query=query, + orm_model=source_table, + skip_archive=True, + session=session, + batch_size=None, + ) + + assert exc_info.value is delete_failure + session.rollback.assert_called_once_with() + session.connection.assert_called_once_with() + assert session.get_bind.call_count == 1 + drop_mock.assert_called_once_with(bind=session.connection.return_value) + + rollback_call_index = tracker.mock_calls.index(call.session.rollback()) + drop_call_index = tracker.mock_calls.index(call.drop(bind=session.connection.return_value)) + commit_call_index = tracker.mock_calls.index(call.session.commit(), drop_call_index) + assert rollback_call_index < drop_call_index < commit_call_index + + @pytest.mark.parametrize( + ("skip_archive", "expected_commit_count"), + [pytest.param(True, 3, id="skip_archive"), pytest.param(False, 2, id="keep_archive")], + ) + def test_do_delete_success_does_not_call_rollback(self, skip_archive, expected_commit_count): + session = MagicMock(spec=Session) + session.get_bind.return_value.dialect.name = "mysql" + session.connection.return_value = object() + session.scalars.return_value.one.side_effect = [1, 0] + session.execute.side_effect = [None, None, None] + + metadata, source_table, target_table, query = _build_do_delete_test_objects() + + with ( + patch("airflow.utils.db_cleanup.reflect_tables", return_value=metadata), + patch("airflow.utils.db_cleanup.timezone.utcnow", return_value=_delete_test_timestamp()), + patch.object(target_table, "drop") as drop_mock, + ): + _do_delete( + query=query, + orm_model=source_table, + skip_archive=skip_archive, + session=session, + batch_size=None, + ) + + session.rollback.assert_not_called() + assert session.commit.call_count == expected_commit_count + if skip_archive: + session.connection.assert_called_once_with() + drop_mock.assert_called_once_with(bind=session.connection.return_value) + else: + session.connection.assert_not_called() + drop_mock.assert_not_called() + @patch("airflow.utils.db.reflect_tables") def test_skip_archive_failure_will_remove_table(self, reflect_tables_mock): """ @@ -591,6 +668,67 @@ def test_skip_archive_failure_will_remove_table(self, reflect_tables_mock): archived_table_names = _get_archived_table_names(["dag_run"], session) assert len(archived_table_names) == 0 + @pytest.mark.backend("mysql") + def test_db_clean_does_not_deadlock_on_fk_violation_mysql(self): + clean_before_date = _create_pinned_dag_version_cleanup_data( + base_date=pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC")) + ) + timeout_seconds = 15 + finished = threading.Event() + result: dict[str, BaseException] = {} + + def cleanup_in_thread(): + try: + with create_session() as session: + _cleanup_table( + **config_dict["dag_version"].__dict__, + clean_before_timestamp=clean_before_date, + dry_run=False, + session=session, + table_names=["dag_version"], + skip_archive=True, + ) + except BaseException as exc: + result["exception"] = exc + finally: + finished.set() + + worker = threading.Thread(target=cleanup_in_thread, daemon=True) + start_time = time.monotonic() + worker.start() + + assert finished.wait(timeout_seconds), ( + "dag_version cleanup timed out after " + f"{time.monotonic() - start_time:.2f} seconds waiting for the FK violation to surface" + ) + + worker.join(timeout=1) + assert not worker.is_alive() + assert isinstance(result.get("exception"), IntegrityError) + + with create_session() as session: + assert _get_dag_version_archive_table_names(session=session) == [] + + @pytest.mark.backend("postgres") + def test_db_clean_failure_path_does_not_break_postgres(self): + clean_before_date = _create_pinned_dag_version_cleanup_data( + base_date=pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC")) + ) + + with pytest.raises(IntegrityError): + with create_session() as session: + _cleanup_table( + **config_dict["dag_version"].__dict__, + clean_before_timestamp=clean_before_date, + dry_run=False, + session=session, + table_names=["dag_version"], + skip_archive=True, + ) + + with create_session() as session: + assert _get_dag_version_archive_table_names(session=session) == [] + def test_no_models_missing(self): """ 1. Verify that for all tables in `airflow.models`, we either have them enabled in db cleanup, @@ -1115,3 +1253,72 @@ def test_extra_filters_keep_in_flight_rows(self): assert seeded[state] in survivors, f"{state} row should NOT be cleaned up" for state in ("success", "failed"): assert seeded[state] not in survivors, f"{state} row should be cleaned up" + + +def _delete_test_timestamp(): + return pendulum.DateTime(2024, 1, 2, 3, 4, 5, tzinfo=pendulum.timezone("UTC")) + + +def _build_do_delete_test_objects(): + metadata = MetaData() + source_table = Table("dag_version", metadata, Column("id", Integer, primary_key=True)) + target_table = Table( + f"{ARCHIVE_TABLE_PREFIX}{source_table.name}__20240102030405", + metadata, + Column("id", Integer, primary_key=True), + ) + query = select(literal(1).label("id")) + return metadata, source_table, target_table, query + + +def _create_pinned_dag_version_cleanup_data(*, base_date): + bundle_name = f"testing-{uuid4()}" + dag_id = f"test-dag_{uuid4()}" + + with create_session() as session: + session.add(DagBundleModel(name=bundle_name)) + session.flush() + + dag = DAG(dag_id=dag_id) + session.add(DagModel(dag_id=dag_id, bundle_name=bundle_name)) + + old_dag_version = DagVersion( + dag_id=dag_id, + version_number=1, + bundle_name=bundle_name, + created_at=base_date, + last_updated=base_date, + ) + new_dag_version = DagVersion( + dag_id=dag_id, + version_number=2, + bundle_name=bundle_name, + created_at=base_date.add(minutes=1), + last_updated=base_date.add(minutes=1), + ) + session.add_all([old_dag_version, new_dag_version]) + session.flush() + + dag_run = DagRun( + dag_id, + run_id="run-1", + run_type=DagRunType.MANUAL, + start_date=base_date, + ) + task = PythonOperator(task_id="dummy-task", dag=dag, python_callable=print) + ti = create_task_instance(task, run_id=dag_run.run_id, dag_version_id=old_dag_version.id) + ti.dag_id = dag_id + ti.start_date = base_date + + session.add_all([dag_run, ti]) + session.commit() + + return base_date.add(days=1) + + +def _get_dag_version_archive_table_names(*, session): + return [ + table_name + for table_name in _get_archived_table_names(["dag_version"], session) + if table_name.startswith(f"{ARCHIVE_TABLE_PREFIX}dag_version__") + ] From 14311cff5b3743dcbde017ef2c70f1d597554206 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Fri, 3 Jul 2026 01:44:17 +0000 Subject: [PATCH 2/9] Rename newsfragment to match PR number The check-newsfragment-pr-number CI job requires the newsfragment filename to match the PR number, not the linked issue number. --- airflow-core/newsfragments/{66177.bugfix.rst => 66296.bugfix.rst} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename airflow-core/newsfragments/{66177.bugfix.rst => 66296.bugfix.rst} (100%) diff --git a/airflow-core/newsfragments/66177.bugfix.rst b/airflow-core/newsfragments/66296.bugfix.rst similarity index 100% rename from airflow-core/newsfragments/66177.bugfix.rst rename to airflow-core/newsfragments/66296.bugfix.rst From 053b50f99f66f55b42a467dc969a5f4ce7daa29f Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Fri, 3 Jul 2026 01:44:17 +0000 Subject: [PATCH 3/9] Remove newsfragment per review --- airflow-core/newsfragments/66296.bugfix.rst | 1 - 1 file changed, 1 deletion(-) delete mode 100644 airflow-core/newsfragments/66296.bugfix.rst diff --git a/airflow-core/newsfragments/66296.bugfix.rst b/airflow-core/newsfragments/66296.bugfix.rst deleted file mode 100644 index dcf9e601c70d6..0000000000000 --- a/airflow-core/newsfragments/66296.bugfix.rst +++ /dev/null @@ -1 +0,0 @@ -Fix ``airflow db clean --skip-archive`` hanging indefinitely on MySQL when a delete fails due to a foreign-key constraint. The session now rolls back before dropping the temporary archive table, so the command fails fast with the database error instead of blocking on metadata locks. From 8d6f67cbca3fa1d10331b0643260351a6193858a Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Fri, 3 Jul 2026 01:44:17 +0000 Subject: [PATCH 4/9] Suppress rollback errors so the original delete error propagates --- airflow-core/src/airflow/utils/db_cleanup.py | 9 ++++-- .../tests/unit/utils/test_db_cleanup.py | 29 +++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/utils/db_cleanup.py b/airflow-core/src/airflow/utils/db_cleanup.py index 25901a8a50a4e..1af241d01cd3e 100644 --- a/airflow-core/src/airflow/utils/db_cleanup.py +++ b/airflow-core/src/airflow/utils/db_cleanup.py @@ -27,7 +27,7 @@ import logging import os from collections.abc import Generator -from contextlib import contextmanager +from contextlib import contextmanager, suppress from dataclasses import dataclass from types import SimpleNamespace from typing import TYPE_CHECKING, Any @@ -324,7 +324,12 @@ def _do_delete( session.commit() except BaseException: - session.rollback() + # Roll back the failed transaction so its locks are released before + # the archive table is dropped in the ``finally`` block below. + # ``rollback()`` itself can raise (e.g. the connection died); suppress + # it so it does not shadow the original error being re-raised. + with suppress(Exception): + session.rollback() raise finally: if target_table is not None and skip_archive: diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index 050d034c2dcc8..efd809ee6d056 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -600,6 +600,35 @@ def test_do_delete_rolls_back_before_drop_on_failure(self): commit_call_index = tracker.mock_calls.index(call.session.commit(), drop_call_index) assert rollback_call_index < drop_call_index < commit_call_index + def test_do_delete_propagates_original_error_when_rollback_fails(self): + session = MagicMock(spec=Session) + session.get_bind.return_value.dialect.name = "mysql" + session.connection.return_value = object() + session.scalars.return_value.one.side_effect = [1, 0] + + metadata, source_table, target_table, query = _build_do_delete_test_objects() + delete_failure = IntegrityError("DELETE FROM dag_version", {}, Exception("fk violation")) + session.execute.side_effect = [None, None, delete_failure] + session.rollback.side_effect = OperationalError("ROLLBACK", {}, Exception("connection lost")) + + with ( + patch("airflow.utils.db_cleanup.reflect_tables", return_value=metadata), + patch("airflow.utils.db_cleanup.timezone.utcnow", return_value=_delete_test_timestamp()), + patch.object(target_table, "drop") as drop_mock, + ): + with pytest.raises(IntegrityError) as exc_info: + _do_delete( + query=query, + orm_model=source_table, + skip_archive=True, + session=session, + batch_size=None, + ) + + assert exc_info.value is delete_failure + session.rollback.assert_called_once_with() + drop_mock.assert_called_once_with(bind=session.connection.return_value) + @pytest.mark.parametrize( ("skip_archive", "expected_commit_count"), [pytest.param(True, 3, id="skip_archive"), pytest.param(False, 2, id="keep_archive")], From eba2b736113afb1c4dfa25dd124171bf559d77e6 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Fri, 3 Jul 2026 01:44:17 +0000 Subject: [PATCH 5/9] Explain why the archive table drop reuses the session connection --- airflow-core/src/airflow/utils/db_cleanup.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/airflow-core/src/airflow/utils/db_cleanup.py b/airflow-core/src/airflow/utils/db_cleanup.py index 1af241d01cd3e..6ba9f4b4481e2 100644 --- a/airflow-core/src/airflow/utils/db_cleanup.py +++ b/airflow-core/src/airflow/utils/db_cleanup.py @@ -333,6 +333,12 @@ def _do_delete( raise finally: if target_table is not None and skip_archive: + # Drop the archive table on the session's own connection. Binding + # the drop to ``session.get_bind()`` (the Engine) would check out a + # *second* pooled connection, and on MySQL its ``DROP TABLE`` blocks + # indefinitely on the metadata lock still held by this session's + # open transaction when the DELETE above failed -- the ``db clean`` + # hang reported in #66177. target_table.drop(bind=session.connection()) session.commit() From ce1cfe1fd9da4a0057cf01b0e09b9cdb55b97f87 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Fri, 3 Jul 2026 01:44:17 +0000 Subject: [PATCH 6/9] Force the FK-failure path in backend regression tests via config copy Keeps the hang-regression tests meaningful once dag_version cleanup learns to skip FK-pinned rows (apache/airflow#68339). --- .../tests/unit/utils/test_db_cleanup.py | 21 +++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index efd809ee6d056..8871437f4ba82 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -710,7 +710,7 @@ def cleanup_in_thread(): try: with create_session() as session: _cleanup_table( - **config_dict["dag_version"].__dict__, + **_dag_version_config_without_row_exclusion(), clean_before_timestamp=clean_before_date, dry_run=False, session=session, @@ -747,7 +747,7 @@ def test_db_clean_failure_path_does_not_break_postgres(self): with pytest.raises(IntegrityError): with create_session() as session: _cleanup_table( - **config_dict["dag_version"].__dict__, + **_dag_version_config_without_row_exclusion(), clean_before_timestamp=clean_before_date, dry_run=False, session=session, @@ -1351,3 +1351,20 @@ def _get_dag_version_archive_table_names(*, session): for table_name in _get_archived_table_names(["dag_version"], session) if table_name.startswith(f"{ARCHIVE_TABLE_PREFIX}dag_version__") ] + + +def _dag_version_config_without_row_exclusion(): + """Live dag_version cleanup config with row-exclusion filters disabled. + + The backend regression tests above pin ``_do_delete``'s rollback-before-drop + behaviour (the MySQL metadata-lock hang from #66177), which requires the + DELETE to actually hit the ``task_instance.dag_version_id`` FK violation. + The live config may exclude FK-pinned rows from the deletion query (see + PR #68339), which would turn these tests into no-ops -- so strip any such + filters from a copy of the config. + """ + config = dict(config_dict["dag_version"].__dict__) + for key in ("extra_filters", "skip_if_referenced"): + if key in config: + config[key] = None + return config From 5a98dd5bc9f382cab3a2b90cb2cc9784a169f72e Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Mon, 6 Jul 2026 03:51:06 +0000 Subject: [PATCH 7/9] Guard archive-table cleanup so it can't mask the original delete error When _do_delete unwinds from a failed DELETE, dropping the archive table in the finally block could itself raise (e.g. a dead connection on MySQL). Python makes a finally-raised error the top-level exception, so that cleanup error would replace the original delete failure -- and an OperationalError from the drop could then be downgraded to a warning by _suppress_with_logging, silently masking the real IntegrityError. Guard the cleanup so it only propagates on the success path; on the failure path it is logged and the original error survives. --- airflow-core/src/airflow/utils/db_cleanup.py | 25 +++++++- .../tests/unit/utils/test_db_cleanup.py | 59 +++++++++++++++++++ 2 files changed, 82 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/utils/db_cleanup.py b/airflow-core/src/airflow/utils/db_cleanup.py index 6ba9f4b4481e2..6deb4e78f6856 100644 --- a/airflow-core/src/airflow/utils/db_cleanup.py +++ b/airflow-core/src/airflow/utils/db_cleanup.py @@ -286,6 +286,10 @@ def _do_delete( target_table_name = f"{ARCHIVE_TABLE_PREFIX}{orm_model.name}__{timestamp_str}{suffix}" print(f"Moving data to table {target_table_name}") target_table = None + # Lets the ``finally`` cleanup below tell the failure path (don't let a + # cleanup error mask the original) from the success path (a cleanup error + # is a real problem and must propagate). + error_raised = False try: if dialect_name == "mysql": @@ -324,6 +328,7 @@ def _do_delete( session.commit() except BaseException: + error_raised = True # Roll back the failed transaction so its locks are released before # the archive table is dropped in the ``finally`` block below. # ``rollback()`` itself can raise (e.g. the connection died); suppress @@ -339,8 +344,24 @@ def _do_delete( # indefinitely on the metadata lock still held by this session's # open transaction when the DELETE above failed -- the ``db clean`` # hang reported in #66177. - target_table.drop(bind=session.connection()) - session.commit() + try: + target_table.drop(bind=session.connection()) + session.commit() + except Exception: + # If we are already unwinding from a delete failure, a cleanup + # error here must not replace the original exception (Python + # makes a ``finally``-raised error the top-level one). Log and + # let the original delete error keep propagating. On the success + # path (no delete error), a drop/commit failure is a real + # problem, so re-raise it. + if not error_raised: + raise + logger.warning( + "Failed to drop archive table %s while cleaning up after a " + "delete failure; propagating the original delete error instead.", + target_table_name, + exc_info=True, + ) print("Finished Performing Delete") diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index 8871437f4ba82..b1f0a14a1d552 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -664,6 +664,65 @@ def test_do_delete_success_does_not_call_rollback(self, skip_archive, expected_c session.connection.assert_not_called() drop_mock.assert_not_called() + def test_do_delete_original_error_survives_archive_drop_failure(self): + """On the failure path, a drop/commit error in the finally block must not + replace the original delete error (nailo2c review, #66296).""" + session = MagicMock(spec=Session) + session.get_bind.return_value.dialect.name = "mysql" + session.connection.return_value = object() + session.scalars.return_value.one.side_effect = [1, 0] + + metadata, source_table, target_table, query = _build_do_delete_test_objects() + delete_failure = IntegrityError("DELETE FROM dag_version", {}, Exception("fk violation")) + session.execute.side_effect = [None, None, delete_failure] + drop_failure = OperationalError("DROP TABLE", {}, Exception("server has gone away")) + + with ( + patch("airflow.utils.db_cleanup.reflect_tables", return_value=metadata), + patch("airflow.utils.db_cleanup.timezone.utcnow", return_value=_delete_test_timestamp()), + patch.object(target_table, "drop", side_effect=drop_failure) as drop_mock, + ): + with pytest.raises(IntegrityError) as exc_info: + _do_delete( + query=query, + orm_model=source_table, + skip_archive=True, + session=session, + batch_size=None, + ) + + assert exc_info.value is delete_failure + drop_mock.assert_called_once_with(bind=session.connection.return_value) + + def test_do_delete_success_propagates_archive_drop_error(self): + """On the success path, a drop/commit failure is a real error and must + still surface (the failure-path guard must not swallow it).""" + session = MagicMock(spec=Session) + session.get_bind.return_value.dialect.name = "mysql" + session.connection.return_value = object() + session.scalars.return_value.one.side_effect = [1, 0] + session.execute.side_effect = [None, None, None] + + metadata, source_table, target_table, query = _build_do_delete_test_objects() + drop_failure = OperationalError("DROP TABLE", {}, Exception("disk full")) + + with ( + patch("airflow.utils.db_cleanup.reflect_tables", return_value=metadata), + patch("airflow.utils.db_cleanup.timezone.utcnow", return_value=_delete_test_timestamp()), + patch.object(target_table, "drop", side_effect=drop_failure), + ): + with pytest.raises(OperationalError) as exc_info: + _do_delete( + query=query, + orm_model=source_table, + skip_archive=True, + session=session, + batch_size=None, + ) + + assert exc_info.value is drop_failure + session.rollback.assert_not_called() + @patch("airflow.utils.db.reflect_tables") def test_skip_archive_failure_will_remove_table(self, reflect_tables_mock): """ From 7b35686cac5c357e5e5059a90ac01e1029ade8f6 Mon Sep 17 00:00:00 2001 From: deepinsight coder <32898216+Vamsi-klu@users.noreply.github.com> Date: Mon, 13 Jul 2026 09:15:17 -0700 Subject: [PATCH 8/9] Update airflow-core/tests/unit/utils/test_db_cleanup.py Co-authored-by: Ephraim Anierobi --- airflow-core/tests/unit/utils/test_db_cleanup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index b1f0a14a1d552..8271d38968b98 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -1416,7 +1416,7 @@ def _dag_version_config_without_row_exclusion(): """Live dag_version cleanup config with row-exclusion filters disabled. The backend regression tests above pin ``_do_delete``'s rollback-before-drop - behaviour (the MySQL metadata-lock hang from #66177), which requires the + behaviour, which requires the DELETE to actually hit the ``task_instance.dag_version_id`` FK violation. The live config may exclude FK-pinned rows from the deletion query (see PR #68339), which would turn these tests into no-ops -- so strip any such From 247d707a00c62dbdfe1a5f87e5619a107ac0c6e8 Mon Sep 17 00:00:00 2001 From: Ramachandra Nalam Date: Mon, 13 Jul 2026 11:03:03 -0700 Subject: [PATCH 9/9] Retry CI after provider compat startup failure The previous run failed because the Mongo provider test fixture could not connect to its temporary mongod container. No source change is needed; this commit retriggers CI so the PR can use a fresh signal.