This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 16715d605b8 Fix airflow db clean never purging the callback table
(#70923) (#72899)
16715d605b8 is described below
commit 16715d605b88250bb26f33356b26ffad1346cae6
Author: Steve Ahn <[email protected]>
AuthorDate: Thu Sep 10 22:38:29 2026 -0700
Fix airflow db clean never purging the callback table (#70923) (#72899)
* Fix airflow db clean never purging the callback table
The table was renamed from callback_request in Airflow 3.2.0 and the
cleanup configuration was not updated with it. A configured table that
does not exist is skipped with only a warning, so these rows were never
deleted and the table grew without bound, with no way for an operator to
purge it.
Only callbacks that can no longer run are removed. A callback still
awaiting execution owns its deadline row through an ON DELETE CASCADE
foreign key, so deleting one would silently drop a deadline that has not
fired yet.
The sla_miss entry is dropped as well, since that table no longer exists
in Airflow 3.
* Purge orphaned partitioned_asset_key_log rows and restore the db clean
coverage check
The check that is meant to catch a metadata table being left out of
airflow db clean looked for the models package at a path relative to the
test file. That path stopped resolving when the sources moved under
airflow-core, so the check examined no models at all and its assertions
held trivially for an empty set. It has not been able to report a missing
table since, which is how the callback table went unpurged for several
releases.
Walking the package's own search path instead keeps the check working
wherever the sources live, and asserting that models were found stops it
from passing on an empty set again.
With the check restored, partitioned_asset_key_log is the one table it
reports that genuinely has no way to be purged: it carries no foreign
key, so its rows are left behind when the partition Dag run they describe
is cascade-deleted with its dag_run. Only those orphans are deleted --
rows whose partition Dag run still exists are the evidence the scheduler
evaluates to decide when that pending run fires, so they are kept
regardless of age. The other uncovered tables are recorded in the
exclusion list with the reason each is safe.
* Purge orphaned scheduled callbacks while keeping unfired deadlines in db
clean
Every callback starts in SCHEDULED, a state outside both the active and the
terminal sets, and an unfired deadline's callback stays there until the
deadline
is missed. Deny-listing active states would therefore purge those callbacks
and,
through the ON DELETE CASCADE foreign key, silently drop deadlines that
have not
fired. The finished states stay allow-listed, so an unknown state keeps its
rows.
A SCHEDULED callback is purged only once no deadline references it:
deleting a
Dag run cascades away its deadline at the database level, which leaves the
callback behind forever.
(cherry picked from commit 478745af84d5e8418271ff4d5cba39408ed74c4e)
---
airflow-core/newsfragments/70923.bugfix.rst | 1 +
airflow-core/src/airflow/utils/db_cleanup.py | 49 +++-
airflow-core/tests/unit/utils/test_db_cleanup.py | 316 ++++++++++++++++++++++-
3 files changed, 355 insertions(+), 11 deletions(-)
diff --git a/airflow-core/newsfragments/70923.bugfix.rst
b/airflow-core/newsfragments/70923.bugfix.rst
new file mode 100644
index 00000000000..b1545b7fb77
--- /dev/null
+++ b/airflow-core/newsfragments/70923.bugfix.rst
@@ -0,0 +1 @@
+Fix ``airflow db clean`` never purging the ``callback`` table. The table was
renamed from ``callback_request`` in Airflow 3.2.0 but the cleanup
configuration kept the old name, and a configured table that does not exist is
skipped with only a warning, so the rows were never deleted and the table grew
without bound. Only callbacks that can no longer run are purged: a callback
still awaiting execution owns its ``deadline`` row through an ``ON DELETE
CASCADE`` foreign key, so deleting one w [...]
diff --git a/airflow-core/src/airflow/utils/db_cleanup.py
b/airflow-core/src/airflow/utils/db_cleanup.py
index 8197d56c42d..dfd0ef9804a 100644
--- a/airflow-core/src/airflow/utils/db_cleanup.py
+++ b/airflow-core/src/airflow/utils/db_cleanup.py
@@ -32,7 +32,7 @@ from dataclasses import dataclass
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
-from sqlalchemy import and_, column, func, inspect, literal, select, table,
text
+from sqlalchemy import and_, column, func, inspect, literal, literal_column,
or_, select, table, text
from sqlalchemy.exc import OperationalError, ProgrammingError
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import aliased
@@ -42,9 +42,11 @@ from airflow._shared.timezones import timezone
from airflow.cli.simple_table import AirflowConsole
from airflow.configuration import conf
from airflow.exceptions import AirflowException
+from airflow.models.callback import TERMINAL_STATES
from airflow.utils.db import reflect_tables
from airflow.utils.helpers import ask_yesno
from airflow.utils.session import NEW_SESSION, provide_session
+from airflow.utils.state import CallbackState
from airflow.utils.types import DagRunType
if TYPE_CHECKING:
@@ -57,6 +59,8 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
ARCHIVE_TABLE_PREFIX = "_airflow_deleted__"
+# Alias _build_query gives the table being cleaned; a correlated extra_filter
refers to it by name.
+_BASE_TABLE_ALIAS = "base"
# Archived tables created by DB migrations
ARCHIVED_TABLES_FROM_DB_MIGRATIONS = [
"_xcom_archive" # Table created by the AF 2 -> 3.0.0 migration when the
XComs had pickled values
@@ -161,9 +165,22 @@ config_list: list[_TableConfig] = [
dependent_tables=["task_instance", "task_state_store", "deadline"],
),
_TableConfig(table_name="asset_event", recency_column_name="timestamp",
dag_id_column_name="dag_id"),
+ # Carries no foreign key, so rows are left behind when the partition Dag
run they describe
+ # is cascade-deleted with its dag_run. Only such orphans may be purged:
rows whose partition
+ # Dag run still exists are the evidence the scheduler evaluates to decide
when that pending
+ # run fires, so age alone must not delete them.
+ _TableConfig(
+ table_name="partitioned_asset_key_log",
+ recency_column_name="created_at",
+ extra_columns=["asset_partition_dag_run_id"],
+ extra_filters=[
+ column("asset_partition_dag_run_id").not_in(
+
select(column("id")).select_from(table("asset_partition_dag_run"))
+ )
+ ],
+ ),
_TableConfig(table_name="import_error", recency_column_name="timestamp"),
_TableConfig(table_name="log", recency_column_name="dttm",
dag_id_column_name="dag_id"),
- _TableConfig(table_name="sla_miss", recency_column_name="timestamp",
dag_id_column_name="dag_id"),
_TableConfig(
table_name="task_instance",
recency_column_name="start_date",
@@ -181,7 +198,31 @@ config_list: list[_TableConfig] = [
_TableConfig(table_name="task_reschedule",
recency_column_name="start_date", dag_id_column_name="dag_id"),
_TableConfig(table_name="xcom", recency_column_name="timestamp",
dag_id_column_name="dag_id"),
_TableConfig(table_name="_xcom_archive", recency_column_name="timestamp",
dag_id_column_name="dag_id"),
- _TableConfig(table_name="callback_request",
recency_column_name="created_at"),
+ _TableConfig(
+ table_name="callback",
+ recency_column_name="created_at",
+ extra_columns=["id", "state"],
+ # Purging a callback cascades to its deadline row, so only finished
callbacks are purged;
+ # a state this code does not know keeps its rows. An unfired
deadline's callback sits in
+ # SCHEDULED, which is neither active nor terminal, until the deadline
is missed; it is
+ # purged only once no deadline references it, as deleting a Dag run
cascades away the
+ # deadline at the database level and leaves the callback behind.
Dag-processor callbacks
+ # carry no state and are deleted as they are dispatched.
+ extra_filters=[
+ or_(
+ column("state").in_(sorted(TERMINAL_STATES)),
+ column("state").is_(None),
+ and_(
+ column("state") == CallbackState.SCHEDULED,
+ ~select(literal(1))
+ .select_from(table("deadline", column("callback_id")))
+ .where(column("callback_id") ==
literal_column(f"{_BASE_TABLE_ALIAS}.id"))
+ .exists(),
+ ),
+ )
+ ],
+ dependent_tables=["deadline"],
+ ),
_TableConfig(table_name="celery_taskmeta",
recency_column_name="date_done"),
_TableConfig(table_name="celery_tasksetmeta",
recency_column_name="date_done"),
_TableConfig(
@@ -385,7 +426,7 @@ def _build_query(
referenced_pk_column: str = "id",
**kwargs,
) -> Select:
- base_table_alias = "base"
+ base_table_alias = _BASE_TABLE_ALIAS
base_table = aliased(orm_model, name=base_table_alias)
query = select(text(f"{base_table_alias}.*")).select_from(base_table)
base_table_recency_col = base_table.c[recency_column.name]
diff --git a/airflow-core/tests/unit/utils/test_db_cleanup.py
b/airflow-core/tests/unit/utils/test_db_cleanup.py
index 4d66a96b1a3..0e61c37f6f0 100644
--- a/airflow-core/tests/unit/utils/test_db_cleanup.py
+++ b/airflow-core/tests/unit/utils/test_db_cleanup.py
@@ -20,13 +20,12 @@ from __future__ import annotations
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 uuid import uuid4
import pendulum
import pytest
-from sqlalchemy import func, inspect, select, text
+from sqlalchemy import func, insert, inspect, select, text
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from sqlalchemy.ext.declarative import DeclarativeMeta
@@ -47,6 +46,7 @@ from airflow.utils.db_cleanup import (
_cleanup_table,
_confirm_drop_archives,
_dump_table_to_file,
+ _effective_table_names,
_get_archived_table_names,
_TableConfig,
config_dict,
@@ -599,11 +599,12 @@ class TestDBCleanup:
"""
import pkgutil
- proj_root = Path(__file__).parents[2].resolve()
- mods = list(
- f"airflow.models.{name}"
- for _, name, _ in pkgutil.iter_modules([str(proj_root /
"airflow/models")])
- )
+ import airflow.models
+
+ # Walk the package's own __path__ rather than rebuilding it from this
file's
+ # location: a path guessed relative to the test resolves to nothing
once the
+ # sources move, leaving the assertions below with an empty set to
check.
+ mods = [f"airflow.models.{name}" for _, name, _ in
pkgutil.iter_modules(airflow.models.__path__)]
all_models = {}
for mod_name in mods:
@@ -643,12 +644,30 @@ class TestDBCleanup:
"dag_priority_parsing_request", # Records are purged once per DAG
Processing loop, not a
# significant source of data.
"dag_bundle", # leave alone - not appropriate for cleanup
+ "team", # leave alone - team configuration, not run data
+ # leave alone - per-asset key/value state, upserted in place (PK
is asset_id+key),
+ # so it is bounded and current, not accumulating history; removed
with its asset
+ "asset_state_store",
+ # Purged indirectly: each of these hangs off a cleaned table by an
+ # ON DELETE CASCADE foreign key, so the rows go when the parent
does.
+ # cascade from dag_run once the partition run has fired; while it
is still
+ # pending its created_dag_run_id is NULL, so those rows are not
cleaned
+ "asset_partition_dag_run",
+ "asset_watcher", # cascade from trigger
+ "dag_favorite", # cascade from dag
+ "deadline_alert", # cascade from serialized_dag, which cascades
from dag_version
+ "hitl_detail", # cascade from task_instance
+ "hitl_detail_history", # cascade from task_instance_history
+ "task_inlet_asset_reference", # cascade from dag
}
from airflow.utils.db_cleanup import config_dict
print(f"all_models={set(all_models)}")
print(f"excl+conf={exclusion_list.union(config_dict)}")
+ # Without this the two assertions below hold trivially for an empty
set,
+ # which is how a table can go unnoticed by this check for several
releases.
+ assert all_models, "discovered no models, so this check would pass
vacuously"
assert set(all_models) - exclusion_list.union(config_dict) == set()
assert exclusion_list.isdisjoint(config_dict)
@@ -1115,3 +1134,286 @@ class TestConnectionTestRequestCleanup:
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"
+
+
+class TestCallbackCleanupConfig:
+ """The callback table is registered under the name it actually has in the
schema."""
+
+ def test_callback_is_configured_under_its_current_name(self):
+ # The table was renamed from callback_request to callback; a config
entry naming a
+ # table that does not exist is skipped with a warning, so the rows are
never purged.
+ assert "callback" in config_dict
+ assert "callback_request" not in config_dict
+
+ def test_removed_tables_are_not_configured(self):
+ assert "sla_miss" not in config_dict
+
+ def test_cleaning_callback_pulls_in_its_dependent_deadline_rows(self):
+ # deadline.callback_id cascades from callback, so deadline has to be
cleaned
+ # (and archived) first, or those rows would vanish unrecorded.
+ selected, _ = _effective_table_names(table_names=["callback"])
+ assert selected == ["deadline", "callback"]
+
+
[email protected]_test
+class TestCallbackCleanup:
+ """Cleanup must purge finished callbacks without disturbing ones that can
still run."""
+
+ def setup_method(self):
+ from tests_common.test_utils.db import clear_db_callbacks,
clear_db_deadline
+
+ clear_db_deadline()
+ clear_db_callbacks()
+ with create_session() as session:
+ for name in _get_archived_table_names(["callback", "deadline"],
session):
+ session.execute(text(f"DROP TABLE {name}"))
+ session.commit()
+
+ def teardown_method(self):
+ from tests_common.test_utils.db import clear_db_callbacks,
clear_db_deadline
+
+ clear_db_deadline()
+ clear_db_callbacks()
+ with create_session() as session:
+ for name in _get_archived_table_names(["callback", "deadline"],
session):
+ session.execute(text(f"DROP TABLE {name}"))
+ session.commit()
+
+ @staticmethod
+ def _add_callback(session, state, created_at):
+ from airflow.executors.workloads.callback import CallbackFetchMethod
+ from airflow.models.callback import Callback
+
+ callback = Callback(priority_weight=1)
+ callback.fetch_method = CallbackFetchMethod.IMPORT_PATH
+ callback.state = state
+ callback.created_at = created_at
+ session.add(callback)
+ session.flush()
+ return callback.id
+
+ @staticmethod
+ def _count_callbacks(session, callback_id):
+ from airflow.models.callback import Callback
+
+ return
session.scalar(select(func.count()).select_from(Callback).where(Callback.id ==
callback_id))
+
+ @staticmethod
+ def _clean_callbacks(cutoff):
+ with create_session() as session:
+ _cleanup_table(
+ **config_dict["callback"].__dict__,
+ clean_before_timestamp=cutoff,
+ dry_run=False,
+ verbose=False,
+ confirm=False,
+ skip_archive=True,
+ session=session,
+ )
+
+ @pytest.mark.parametrize(
+ ("state", "should_survive"),
+ [
+ ("success", False),
+ ("failed", False),
+ (None, False),
+ ("scheduled", False),
+ ("pending", True),
+ ("queued", True),
+ ("running", True),
+ ],
+ ids=[
+ "success",
+ "failed",
+ "dag-processor-null-state",
+ "scheduled-orphan",
+ "pending",
+ "queued",
+ "running",
+ ],
+ )
+ def test_only_finished_callbacks_are_purged(self, state, should_survive):
+ old = pendulum.now(tz="UTC").subtract(days=30)
+ cutoff = pendulum.now(tz="UTC").subtract(days=1)
+
+ with create_session() as session:
+ callback_id = self._add_callback(session, state, old)
+ session.commit()
+
+ self._clean_callbacks(cutoff)
+
+ with create_session() as session:
+ survived = self._count_callbacks(session, callback_id)
+
+ assert bool(survived) is should_survive
+
+ def test_unfired_deadline_callback_and_its_deadline_survive(self,
dag_maker):
+ from airflow.models.deadline import Deadline
+ from airflow.sdk.definitions.callback import AsyncCallback
+ from airflow.utils.state import CallbackState
+
+ old = pendulum.now(tz="UTC").subtract(days=30)
+ cutoff = pendulum.now(tz="UTC").subtract(days=1)
+ with dag_maker("test_db_clean_unfired_deadline"):
+ pass
+ dag_run = dag_maker.create_dagrun()
+
+ with create_session() as session:
+ deadline = Deadline(
+ deadline_time=pendulum.now(tz="UTC").add(days=10),
+
callback=AsyncCallback("tests.unit.models.test_deadline.callback_for_deadline"),
+ dagrun_id=dag_run.id,
+ deadline_alert_id=None,
+ )
+ session.add(deadline)
+ session.flush()
+ deadline.callback.created_at = old
+ assert deadline.callback.state == CallbackState.SCHEDULED
+ deadline_id, callback_id = deadline.id, deadline.callback.id
+
+ self._clean_callbacks(cutoff)
+
+ with create_session() as session:
+ assert self._count_callbacks(session, callback_id) == 1
+ assert (
+
session.scalar(select(func.count()).select_from(Deadline).where(Deadline.id ==
deadline_id))
+ == 1
+ )
+
+ def test_recent_finished_callback_is_kept(self):
+ recent = pendulum.now(tz="UTC")
+ cutoff = pendulum.now(tz="UTC").subtract(days=1)
+
+ with create_session() as session:
+ callback_id = self._add_callback(session, "success", recent)
+ session.commit()
+
+ self._clean_callbacks(cutoff)
+
+ with create_session() as session:
+ assert self._count_callbacks(session, callback_id) == 1
+
+ def test_finished_callbacks_are_archived_not_just_deleted(self):
+ """Archiving is the default path, so the purged rows must be
recoverable."""
+ from airflow.models.callback import Callback
+
+ old = pendulum.now(tz="UTC").subtract(days=30)
+ cutoff = pendulum.now(tz="UTC").subtract(days=1)
+
+ with create_session() as session:
+ self._add_callback(session, "success", old)
+ session.commit()
+
+ with create_session() as session:
+ _cleanup_table(
+ **config_dict["callback"].__dict__,
+ clean_before_timestamp=cutoff,
+ dry_run=False,
+ verbose=False,
+ confirm=False,
+ skip_archive=False,
+ session=session,
+ )
+
+ with create_session() as session:
+ archives = _get_archived_table_names(["callback"], session)
+ assert archives, "no archive table was created"
+ archived = sum(
+ session.execute(text(f"SELECT count(*) FROM {name}")).scalar()
for name in archives
+ )
+ assert archived == 1
+ assert session.scalar(select(func.count()).select_from(Callback))
== 0
+
+ def test_deadline_of_a_runnable_callback_is_not_cascade_deleted(self):
+ """deadline.callback_id is ON DELETE CASCADE, so purging a live
callback would drop its deadline."""
+ old = pendulum.now(tz="UTC").subtract(days=30)
+ cutoff = pendulum.now(tz="UTC").subtract(days=1)
+ future = pendulum.now(tz="UTC").add(days=365)
+
+ from airflow.models.deadline import Deadline
+
+ with create_session() as session:
+ callback_id = self._add_callback(session, "scheduled", old)
+ session.execute(
+ insert(Deadline.__table__).values(
+ id=uuid4(),
+ deadline_time=future,
+ callback_id=callback_id,
+ created_at=old,
+ last_updated_at=old,
+ missed=False,
+ )
+ )
+ session.commit()
+
+ self._clean_callbacks(cutoff)
+
+ with create_session() as session:
+ assert session.scalar(select(func.count()).select_from(Deadline))
== 1
+
+
[email protected]_test
+class TestPartitionedAssetKeyLogCleanup:
+ """Only orphaned key-log rows may go: live ones drive pending
partition-run evaluation."""
+
+ def setup_method(self):
+ with create_session() as session:
+ session.execute(text("DELETE FROM partitioned_asset_key_log"))
+ session.execute(text("DELETE FROM asset_partition_dag_run"))
+ session.commit()
+
+ teardown_method = setup_method
+
+ @staticmethod
+ def _add_key_log(session, apdr_id, created_at):
+ from airflow.models.asset import PartitionedAssetKeyLog
+
+ row = PartitionedAssetKeyLog(
+ asset_id=1,
+ asset_event_id=1,
+ asset_partition_dag_run_id=apdr_id,
+ source_partition_key="2024-01-01",
+ target_dag_id="cleanup_probe_dag",
+ target_partition_key="2024-01-01",
+ )
+ row.created_at = created_at
+ session.add(row)
+ session.flush()
+ return row.id
+
+ def test_only_orphaned_key_log_rows_are_purged(self):
+ from airflow.models.asset import AssetPartitionDagRun,
PartitionedAssetKeyLog
+
+ old = pendulum.now(tz="UTC").subtract(days=30)
+ cutoff = pendulum.now(tz="UTC").subtract(days=1)
+
+ with create_session() as session:
+ apdr = AssetPartitionDagRun(target_dag_id="cleanup_probe_dag",
partition_key="2024-01-01")
+ session.add(apdr)
+ session.flush()
+ ids = {
+ "old orphan": self._add_key_log(session, apdr_id=apdr.id +
1000, created_at=old),
+ "old but live": self._add_key_log(session, apdr_id=apdr.id,
created_at=old),
+ "recent orphan": self._add_key_log(
+ session, apdr_id=apdr.id + 1000,
created_at=pendulum.now(tz="UTC")
+ ),
+ }
+ session.commit()
+
+ with create_session() as session:
+ _cleanup_table(
+ **config_dict["partitioned_asset_key_log"].__dict__,
+ clean_before_timestamp=cutoff,
+ dry_run=False,
+ verbose=False,
+ confirm=False,
+ skip_archive=True,
+ session=session,
+ )
+
+ with create_session() as session:
+ surviving =
set(session.scalars(select(PartitionedAssetKeyLog.id)).all())
+
+ assert ids["old orphan"] not in surviving
+ assert ids["old but live"] in surviving, "evidence for a pending
partition run must not be deleted"
+ assert ids["recent orphan"] in surviving, "age filter still applies to
orphans"
