From 01407b17e0900d558c00196838c1b49afd331c27 Mon Sep 17 00:00:00 2001 From: mallesh-bot Date: Mon, 21 Sep 2026 00:35:53 +0530 Subject: [PATCH 1/2] Stop git-sync v4 from silently overriding chart's ref value In git-sync v4 the deprecated GIT_SYNC_BRANCH and GIT_SYNC_REV env vars take precedence over GITSYNC_REF, so the chart's git-sync container ignored dags.gitSync.ref (e.g. pinning to a tag like v1.2.3) and kept tracking the branch at rev. Emit GITSYNC_REF only when dags.gitSync.ref is set, and only emit the deprecated GIT_SYNC_BRANCH / GIT_SYNC_REV when ref is unset (v3 mode and the default). Default ref to null so the shipped default render is unchanged for existing v3 and v4 users -- only users who explicitly set ref see behaviour change, and that behaviour now matches what the option documents. closes: #42918 --- chart/newsfragments/42918.bugfix.rst | 11 ++++ chart/templates/_helpers.yaml | 7 ++- .../airflow_aux/test_pod_template_file.py | 2 - .../other/test_git_sync_scheduler.py | 54 +++++++++++++++++-- chart/values.schema.json | 9 ++-- chart/values.yaml | 9 +++- 6 files changed, 79 insertions(+), 13 deletions(-) create mode 100644 chart/newsfragments/42918.bugfix.rst diff --git a/chart/newsfragments/42918.bugfix.rst b/chart/newsfragments/42918.bugfix.rst new file mode 100644 index 0000000000000..afc5c91738b77 --- /dev/null +++ b/chart/newsfragments/42918.bugfix.rst @@ -0,0 +1,11 @@ +Fix ``dags.gitSync.ref`` being silently overridden under git-sync v4 + +In git-sync v4 the deprecated ``GIT_SYNC_BRANCH`` and ``GIT_SYNC_REV`` env vars +override ``GITSYNC_REF``. The chart used to emit all three unconditionally, so +setting ``dags.gitSync.ref: v1.2.3`` (to pin the sidecar to a tag) had no +effect and the container kept tracking ``branch`` at ``rev``. The chart now +emits ``GITSYNC_REF`` only when ``ref`` is set and, in that case, omits the +deprecated pair. The default of ``ref`` has been changed to ``~`` so the +default render is unchanged for existing (v3 and v4) users; only users who +explicitly set ``ref`` see behavior change — and that behavior now matches +what the option documents. diff --git a/chart/templates/_helpers.yaml b/chart/templates/_helpers.yaml index 9a5238fdc8c91..610ccd3978bcf 100644 --- a/chart/templates/_helpers.yaml +++ b/chart/templates/_helpers.yaml @@ -220,12 +220,15 @@ If release name contains chart name it will be used as a full name. {{- end }} envFrom: {{- include "custom_git_sync_environment_from" . | default "\n []" | indent 2 }} env: - - name: GIT_SYNC_REV - value: {{ .Values.dags.gitSync.rev | quote }} + {{- if .Values.dags.gitSync.ref }} - name: GITSYNC_REF value: {{ .Values.dags.gitSync.ref | quote }} + {{- else }} + - name: GIT_SYNC_REV + value: {{ .Values.dags.gitSync.rev | quote }} - name: GIT_SYNC_BRANCH value: {{ .Values.dags.gitSync.branch | quote }} + {{- end }} - name: GIT_SYNC_REPO value: {{ .Values.dags.gitSync.repo | quote }} - name: GITSYNC_REPO diff --git a/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py b/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py index 6897086f7977a..08e0fe25a21d6 100644 --- a/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py +++ b/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py @@ -98,9 +98,7 @@ def test_should_add_an_init_container_if_git_sync_is_true(self): "imagePullPolicy": "Always", "envFrom": [{"secretRef": {"name": "proxy-config"}}], "env": [ - {"name": "GIT_SYNC_REV", "value": "HEAD"}, {"name": "GITSYNC_REF", "value": "test-branch"}, - {"name": "GIT_SYNC_BRANCH", "value": "test-branch"}, {"name": "GIT_SYNC_REPO", "value": "https://github.com/apache/airflow.git"}, {"name": "GITSYNC_REPO", "value": "https://github.com/apache/airflow.git"}, {"name": "GIT_SYNC_DEPTH", "value": "1"}, diff --git a/chart/tests/helm_tests/other/test_git_sync_scheduler.py b/chart/tests/helm_tests/other/test_git_sync_scheduler.py index ebcc6f1d06a1d..5f023a2c2fc5a 100644 --- a/chart/tests/helm_tests/other/test_git_sync_scheduler.py +++ b/chart/tests/helm_tests/other/test_git_sync_scheduler.py @@ -96,9 +96,7 @@ def test_validate_the_git_sync_container_spec(self): "imagePullPolicy": "Always", "envFrom": [{"secretRef": {"name": "proxy-config"}}], "env": [ - {"name": "GIT_SYNC_REV", "value": "HEAD"}, {"name": "GITSYNC_REF", "value": "test-branch"}, - {"name": "GIT_SYNC_BRANCH", "value": "test-branch"}, {"name": "GIT_SYNC_REPO", "value": "https://github.com/apache/airflow.git"}, {"name": "GITSYNC_REPO", "value": "https://github.com/apache/airflow.git"}, {"name": "GIT_SYNC_DEPTH", "value": "1"}, @@ -168,9 +166,7 @@ def test_validate_the_git_sync_container_spec_if_wait_specified(self): "imagePullPolicy": "Always", "envFrom": [{"secretRef": {"name": "proxy-config"}}], "env": [ - {"name": "GIT_SYNC_REV", "value": "HEAD"}, {"name": "GITSYNC_REF", "value": "test-branch"}, - {"name": "GIT_SYNC_BRANCH", "value": "test-branch"}, {"name": "GIT_SYNC_REPO", "value": "https://github.com/apache/airflow.git"}, {"name": "GITSYNC_REPO", "value": "https://github.com/apache/airflow.git"}, {"name": "GIT_SYNC_DEPTH", "value": "1"}, @@ -199,6 +195,56 @@ def test_validate_the_git_sync_container_spec_if_wait_specified(self): }, } + def test_ref_takes_precedence_over_deprecated_branch_and_rev(self): + """When ``ref`` is set, only ``GITSYNC_REF`` should be emitted; the + deprecated ``GIT_SYNC_BRANCH`` and ``GIT_SYNC_REV`` would override it + under git-sync v4 (see #42918).""" + docs = render_chart( + values={ + "executor": "LocalExecutor", + "dags": { + "gitSync": { + "enabled": True, + "branch": "main", + "rev": "HEAD", + "ref": "v1.2.3", + } + }, + }, + show_only=["templates/scheduler/scheduler-deployment.yaml"], + ) + + env = jmespath.search("spec.template.spec.containers[1].env", docs[0]) + env_names = {item["name"] for item in env} + assert {"name": "GITSYNC_REF", "value": "v1.2.3"} in env + assert "GIT_SYNC_BRANCH" not in env_names + assert "GIT_SYNC_REV" not in env_names + + def test_deprecated_branch_and_rev_are_emitted_when_ref_is_unset(self): + """Without ``ref`` (git-sync v3 mode, and the shipped default), the + chart still emits ``GIT_SYNC_BRANCH`` and ``GIT_SYNC_REV`` so v3 images + keep working.""" + docs = render_chart( + values={ + "executor": "LocalExecutor", + "dags": { + "gitSync": { + "enabled": True, + "branch": "main", + "rev": "HEAD", + "ref": None, + } + }, + }, + show_only=["templates/scheduler/scheduler-deployment.yaml"], + ) + + env = jmespath.search("spec.template.spec.containers[1].env", docs[0]) + env_names = {item["name"] for item in env} + assert {"name": "GIT_SYNC_BRANCH", "value": "main"} in env + assert {"name": "GIT_SYNC_REV", "value": "HEAD"} in env + assert "GITSYNC_REF" not in env_names + def test_validate_if_ssh_params_are_added(self): docs = render_chart( values={ diff --git a/chart/values.schema.json b/chart/values.schema.json index a0fb7d8379c99..e1ae59a1107b3 100644 --- a/chart/values.schema.json +++ b/chart/values.schema.json @@ -9809,9 +9809,12 @@ "default": "HEAD" }, "ref": { - "description": "Git revision branch, tag, or hash.", - "type": "string", - "default": "v2-2-stable" + "description": "Git revision branch, tag, or hash (git-sync v4 only). When set, the chart sends only GITSYNC_REF and omits the deprecated GIT_SYNC_BRANCH / GIT_SYNC_REV, which would otherwise override GITSYNC_REF in git-sync v4.", + "type": [ + "string", + "null" + ], + "default": null }, "depth": { "description": "Repository depth.", diff --git a/chart/values.yaml b/chart/values.yaml index ca788cc61d501..70bba1d1bb864 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -3523,11 +3523,16 @@ dags: # SSH example: git@github.com:apache/airflow.git # HTTPS example: https://github.com/apache/airflow.git + # For git-sync v3: set ``branch`` and ``rev``. In v4 these two are deprecated + # and, if emitted, override ``ref``, so the chart only sends them when + # ``ref`` is unset. The default keeps them for v3 compatibility. branch: v2-2-stable rev: HEAD - # The git revision (branch, tag, or hash) to check out, v4 only - ref: v2-2-stable + # The git revision (branch, tag, or hash) to check out, v4 only. + # When set, the chart sends only ``GITSYNC_REF`` and omits the deprecated + # ``GIT_SYNC_BRANCH`` / ``GIT_SYNC_REV`` so v4 git-sync does not override it. + ref: ~ depth: 1 From 117ba70e3bd408caa5ce01f5429fb896ff8e00a4 Mon Sep 17 00:00:00 2001 From: mallesh-bot Date: Mon, 21 Sep 2026 00:53:02 +0530 Subject: [PATCH 2/2] Make db clean --skip-archive actually skip the archive table Previously _do_delete always went through a CTAS/INSERT-SELECT into an _airflow_deleted__* archive table and then dropped it in ``finally`` when skip_archive=True. Every row was written twice, and on a deployment with a tight statement timeout the DELETE or DROP could time out, leaving a partial archive behind and failing the cleanup. Add a _delete_directly path that DELETEs the batch's rows in place using an IN-subquery over the same limited_query, per-batch commit preserved. Route skip_archive=True through it and drop the archive-then-drop dance entirely. The archive-then-delete flow is untouched for skip_archive=False (the default), so the existing well-tested path still governs users who want the archive kept. The four mock tests that exercised the archive-drop-on-failure branch under skip_archive=True have been replaced with tests covering the new either/or routing: one asserts skip_archive=True never creates the archive, the other asserts skip_archive=False still uses the archive flow. closes: #42003 --- airflow-core/newsfragments/42003.bugfix.rst | 1 + airflow-core/src/airflow/utils/db_cleanup.py | 179 ++++++++++-------- .../tests/unit/utils/test_db_cleanup.py | 159 +++------------- 3 files changed, 131 insertions(+), 208 deletions(-) create mode 100644 airflow-core/newsfragments/42003.bugfix.rst diff --git a/airflow-core/newsfragments/42003.bugfix.rst b/airflow-core/newsfragments/42003.bugfix.rst new file mode 100644 index 0000000000000..17663d5ac8862 --- /dev/null +++ b/airflow-core/newsfragments/42003.bugfix.rst @@ -0,0 +1 @@ +Fix ``airflow db clean --skip-archive`` still creating (and then dropping) the ``_airflow_deleted__*`` archive table for every batch. The command now deletes matching rows directly from the source table, so each batch does one write instead of a copy-and-drop pair -- avoiding the statement-timeout failures that could leave a partial archive table behind on deployments with tight timeouts. diff --git a/airflow-core/src/airflow/utils/db_cleanup.py b/airflow-core/src/airflow/utils/db_cleanup.py index a6213d7de9e63..da3ca55125027 100644 --- a/airflow-core/src/airflow/utils/db_cleanup.py +++ b/airflow-core/src/airflow/utils/db_cleanup.py @@ -28,7 +28,7 @@ import logging import os from collections.abc import Generator -from contextlib import contextmanager, suppress +from contextlib import contextmanager from types import SimpleNamespace from typing import TYPE_CHECKING, Any @@ -408,6 +408,20 @@ def _do_delete( else: print("Performing Delete...") + if skip_archive: + # The archive-and-drop dance below writes every row twice and, with a low + # statement timeout, can time out before the DROP -- so leave the archive + # out entirely (#42003). Delete the batch's rows in place with a bounded + # IN-subquery against the same limited_query, then commit per batch to + # release row locks. + _delete_directly( + limited_query=limited_query, + source_table_name=source_table_name, + session=session, + dialect_name=dialect_name, + ) + continue + # using bulk delete # create a new table and copy the rows there timestamp_str = re.sub(r"[^\d]", "", timezone.utcnow().isoformat())[:14] @@ -416,87 +430,100 @@ def _do_delete( 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": - # MySQL with replication needs this split into two queries, so just do it for all MySQL - # ERROR 1786 (HY000): Statement violates GTID consistency: CREATE TABLE ... SELECT. - session.execute(text(f"CREATE TABLE {target_table_name} LIKE {source_table_name}")) - metadata = reflect_tables([target_table_name], session) - target_table = metadata.tables[target_table_name] - insert_stm = target_table.insert().from_select(target_table.c, limited_query) - logger.debug("insert statement:\n%s", insert_stm.compile()) - session.execute(insert_stm) - else: - stmt = CreateTableAs(target_table_name, limited_query.selectable) - logger.debug("ctas query:\n%s", stmt.compile()) - session.execute(stmt) - session.commit() - - # delete the rows from the old table - metadata = reflect_tables([source_table_name, target_table_name], session) - source_table = metadata.tables[source_table_name] + + if dialect_name == "mysql": + # MySQL with replication needs this split into two queries, so just do it for all MySQL + # ERROR 1786 (HY000): Statement violates GTID consistency: CREATE TABLE ... SELECT. + session.execute(text(f"CREATE TABLE {target_table_name} LIKE {source_table_name}")) + metadata = reflect_tables([target_table_name], session) target_table = metadata.tables[target_table_name] - logger.debug("rows moved; purging from %s", source_table.name) - if dialect_name == "sqlite": - pk_cols = source_table.primary_key.columns - delete = source_table.delete().where( - tuple_(*pk_cols).in_( - select(*[target_table.c[x.name] for x in source_table.primary_key.columns]) - ) - ) - else: - delete = source_table.delete().where( - and_(*[col == target_table.c[col.name] for col in source_table.primary_key.columns]) + insert_stm = target_table.insert().from_select(target_table.c, limited_query) + logger.debug("insert statement:\n%s", insert_stm.compile()) + session.execute(insert_stm) + else: + stmt = CreateTableAs(target_table_name, limited_query.selectable) + logger.debug("ctas query:\n%s", stmt.compile()) + session.execute(stmt) + session.commit() + + # delete the rows from the old table + metadata = reflect_tables([source_table_name, target_table_name], session) + source_table = metadata.tables[source_table_name] + target_table = metadata.tables[target_table_name] + logger.debug("rows moved; purging from %s", source_table.name) + if dialect_name == "sqlite": + pk_cols = source_table.primary_key.columns + delete = source_table.delete().where( + tuple_(*pk_cols).in_( + select(*[target_table.c[x.name] for x in source_table.primary_key.columns]) ) - logger.debug("delete statement:\n%s", delete.compile()) - session.execute(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 - # 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: - # 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. - 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, - ) + ) + else: + delete = source_table.delete().where( + and_(*[col == target_table.c[col.name] for col in source_table.primary_key.columns]) + ) + logger.debug("delete statement:\n%s", delete.compile()) + session.execute(delete) + session.commit() print("Finished Performing Delete") +def _delete_directly( + *, + limited_query: Select, + source_table_name: str, + session: Session, + dialect_name: str, +) -> None: + """ + DELETE the batch's rows from the source table without an archive copy. + + Used by ``_do_delete`` when ``skip_archive=True``. The IN-subquery references + the ``base``-aliased source table (see ``_build_query``) so the extra filters + that ``_cleanup_table`` layered on ``limited_query`` still apply. + """ + metadata = reflect_tables([source_table_name], session) + source_table = metadata.tables[source_table_name] + pk_cols = list(source_table.primary_key.columns) + if not pk_cols: + raise ValueError(f"Table {source_table_name} has no primary key columns available for cleanup.") + + # ``limited_query`` was built with ``select(text("base.*")).select_from(aliased(orm_model, "base"))``, + # so the columns we want to feed the DELETE come from the ``base`` alias, not from the reflected + # source table itself. + pk_projection = limited_query.with_only_columns( + *[literal_column(f"{_BASE_TABLE_ALIAS}.{col.name}").label(col.name) for col in pk_cols] + ) + rows_to_delete = pk_projection.subquery("rows_to_delete") + + if len(pk_cols) == 1: + pk_col = pk_cols[0] + delete = source_table.delete().where( + source_table.c[pk_col.name].in_(select(rows_to_delete.c[pk_col.name])) + ) + elif dialect_name == "sqlite": + # SQLite supports row-value IN, so this path matches the archive-flow's SQLite branch. + delete = source_table.delete().where( + tuple_(*[source_table.c[c.name] for c in pk_cols]).in_( + select(*[rows_to_delete.c[c.name] for c in pk_cols]) + ) + ) + else: + # MySQL 5.7 rejects the composite tuple IN (SELECT ...) form used above, so translate + # the correlated equality join into an EXISTS clause -- the same shape as the archive + # branch's ``and_(col == target.c[col.name])`` join. + delete = source_table.delete().where( + select(literal(1)) + .select_from(rows_to_delete) + .where(and_(*[source_table.c[c.name] == rows_to_delete.c[c.name] for c in pk_cols])) + .exists() + ) + logger.debug("direct delete statement:\n%s", delete.compile()) + session.execute(delete) + session.commit() + + def _subquery_keep_last( *, recency_column, diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py b/airflow-core/tests/unit/utils/test_db_cleanup.py index 01e3ade8ce346..4ce376bfc692d 100644 --- a/airflow-core/tests/unit/utils/test_db_cleanup.py +++ b/airflow-core/tests/unit/utils/test_db_cleanup.py @@ -22,7 +22,7 @@ from contextlib import suppress from importlib import import_module from io import StringIO -from unittest.mock import MagicMock, call, mock_open, patch +from unittest.mock import MagicMock, mock_open, patch from uuid import uuid4 import pendulum @@ -671,166 +671,61 @@ def test_table_config_dag_id_scope_requires_fk_column(self): # "dagrun_id" intentionally omitted from extra_columns ) - def test_do_delete_rolls_back_before_drop_on_failure(self): + def test_do_delete_skip_archive_never_creates_archive_table(self): + """``skip_archive=True`` must go through ``_delete_directly`` and never + touch the ``_airflow_deleted__*`` archive table (#42003).""" 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] + 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, - ): - 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 - - 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")], - ) - 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, + patch("airflow.utils.db_cleanup._delete_directly") as delete_directly_mock, ): _do_delete( query=query, orm_model=source_table, - skip_archive=skip_archive, + skip_archive=True, session=session, batch_size=None, ) + delete_directly_mock.assert_called_once() + kwargs = delete_directly_mock.call_args.kwargs + assert kwargs["source_table_name"] == source_table.name + assert kwargs["dialect_name"] == "mysql" + # No CTAS / INSERT SELECT / DROP executed via session.execute (only the initial COUNT). + session.execute.assert_not_called() 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() - - 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).""" + def test_do_delete_keep_archive_uses_archive_flow(self): + """``skip_archive=False`` keeps the archive-then-delete flow: 2 commits + (archive commit + source-delete commit), no ``_delete_directly`` call.""" 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")) + 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", side_effect=drop_failure), + patch("airflow.utils.db_cleanup._delete_directly") as delete_directly_mock, ): - with pytest.raises(OperationalError) as exc_info: - _do_delete( - query=query, - orm_model=source_table, - skip_archive=True, - session=session, - batch_size=None, - ) + _do_delete( + query=query, + orm_model=source_table, + skip_archive=False, + session=session, + batch_size=None, + ) - assert exc_info.value is drop_failure + delete_directly_mock.assert_not_called() session.rollback.assert_not_called() + assert session.commit.call_count == 2 @patch("airflow.utils.db.reflect_tables") def test_skip_archive_failure_will_remove_table(self, reflect_tables_mock):