This is an automated email from the ASF dual-hosted git repository.
uranusjr 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 6dd0169c8d1 Delete triggers not used by any assets (#54929)
6dd0169c8d1 is described below
commit 6dd0169c8d1b9e78c04e58d4f8385bcf316dffe2
Author: Tzu-ping Chung <[email protected]>
AuthorDate: Thu Aug 28 00:03:42 2025 +0800
Delete triggers not used by any assets (#54929)
---
.../src/airflow/jobs/scheduler_job_runner.py | 20 ++++++-
.../tests/unit/dag_processing/test_collection.py | 3 +-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 69 ++++++++++++++++++----
airflow-core/tests/unit/jobs/test_triggerer_job.py | 3 +-
4 files changed, 79 insertions(+), 16 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 300d712e9d9..20a92efeb86 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -56,6 +56,7 @@ from airflow.models.asset import (
DagScheduleAssetAliasReference,
DagScheduleAssetReference,
TaskOutletAssetReference,
+ asset_trigger_association_table,
)
from airflow.models.backfill import Backfill
from airflow.models.dag import DAG, DagModel
@@ -65,7 +66,7 @@ from airflow.models.dagrun import DagRun
from airflow.models.dagwarning import DagWarning, DagWarningType
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.taskinstance import TaskInstance
-from airflow.models.trigger import TRIGGER_FAIL_REPR, TriggerFailureReason
+from airflow.models.trigger import TRIGGER_FAIL_REPR, Trigger,
TriggerFailureReason
from airflow.stats import Stats
from airflow.ti_deps.dependencies_states import EXECUTION_STATES
from airflow.timetables.simple import AssetTriggeredTimetable
@@ -1233,6 +1234,10 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
conf.getfloat("scheduler", "parsing_cleanup_interval"),
self._update_asset_orphanage,
)
+ timers.call_regular_interval(
+ conf.getfloat("scheduler", "parsing_cleanup_interval"),
+ self._remove_unreferenced_triggers,
+ )
if any(x.is_local for x in self.job.executors):
bundle_cleanup_mgr = BundleUsageTrackingManager()
@@ -2381,6 +2386,19 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
return task_instance_heartbeat_timeout_message_details
+ @provide_session
+ def _remove_unreferenced_triggers(self, *, session: Session = NEW_SESSION)
-> None:
+ """Remove triggers that are no longer used by anything."""
+ session.execute(
+ delete(Trigger)
+ .where(
+
Trigger.id.not_in(select(asset_trigger_association_table.c.trigger_id)),
+ Trigger.id.not_in(select(Deadline.trigger_id)),
+ Trigger.id.not_in(select(TaskInstance.trigger_id)),
+ )
+ .execution_options(synchronize_session="fetch")
+ )
+
@provide_session
def _update_asset_orphanage(self, session: Session = NEW_SESSION) -> None:
"""
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index b253a72a90e..924f7757b33 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -41,7 +41,7 @@ from airflow.dag_processing.collection import (
)
from airflow.exceptions import SerializationError
from airflow.listeners.listener import get_listener_manager
-from airflow.models import DagModel, DagRun, Trigger
+from airflow.models import DagModel, DagRun
from airflow.models.asset import (
AssetActive,
AssetModel,
@@ -165,7 +165,6 @@ class TestAssetModelOperation:
asset_model = session.scalars(select(AssetModel)).one()
assert len(asset_model.triggers) == expected_num_triggers
- assert session.scalar(select(func.count()).select_from(Trigger)) ==
expected_num_triggers
@pytest.mark.parametrize(
"schedule, model, columns, expected",
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 7b4d31e843c..312edb66f9d 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -43,6 +43,7 @@ from airflow.api_fastapi.auth.tokens import JWTGenerator
from airflow.assets.manager import AssetManager
from airflow.callbacks.callback_requests import DagCallbackRequest,
DagRunContext, TaskCallbackRequest
from airflow.callbacks.database_callback_sink import DatabaseCallbackSink
+from airflow.dag_processing.collection import AssetModelOperation,
DagModelOperation
from airflow.exceptions import AirflowException
from airflow.executors.base_executor import BaseExecutor
from airflow.executors.executor_constants import MOCK_EXECUTOR
@@ -50,14 +51,7 @@ from airflow.executors.executor_loader import ExecutorLoader
from airflow.executors.executor_utils import ExecutorName
from airflow.jobs.job import Job, run_job
from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
-from airflow.models import Deadline
-from airflow.models.asset import (
- AssetActive,
- AssetAliasModel,
- AssetDagRunQueue,
- AssetEvent,
- AssetModel,
-)
+from airflow.models.asset import AssetActive, AssetAliasModel,
AssetDagRunQueue, AssetEvent, AssetModel
from airflow.models.backfill import Backfill, _create_backfill
from airflow.models.dag import DAG, DagModel
from airflow.models.dag_version import DagVersion
@@ -65,15 +59,16 @@ from airflow.models.dagbag import DagBag, sync_bag_to_db
from airflow.models.dagrun import DagRun
from airflow.models.dagwarning import DagWarning
from airflow.models.db_callback_request import DbCallbackRequest
-from airflow.models.deadline import DeadlineCallbackState
+from airflow.models.deadline import Deadline, DeadlineCallbackState
from airflow.models.log import Log
from airflow.models.pool import Pool
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.taskinstance import TaskInstance
+from airflow.models.trigger import Trigger
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.empty import EmptyOperator
-from airflow.sdk import task
-from airflow.sdk.definitions.asset import Asset, AssetAlias
+from airflow.providers.standard.triggers.temporal import DateTimeTrigger
+from airflow.sdk import Asset, AssetAlias, AssetWatcher, task
from airflow.sdk.definitions.deadline import AsyncCallback
from airflow.serialization.serialized_objects import LazyDeserializedDAG,
SerializedDAG
from airflow.timetables.base import DataInterval
@@ -98,6 +93,7 @@ from tests_common.test_utils.db import (
clear_db_pools,
clear_db_runs,
clear_db_serialized_dags,
+ clear_db_triggers,
set_default_pool_slots,
)
from tests_common.test_utils.mock_executor import MockExecutor
@@ -196,6 +192,7 @@ class TestSchedulerJob:
clear_db_jobs()
clear_db_assets()
clear_db_deadline()
+ clear_db_triggers()
@pytest.fixture(autouse=True)
def per_test(self) -> Generator:
@@ -6468,6 +6465,56 @@ class TestSchedulerJob:
assert orphaned == [asset1]
assert [asset.updated_at for asset in orphaned] ==
updated_at_timestamps
+ @pytest.mark.parametrize(
+ "paused, stale, expected_classpath",
+ [
+ pytest.param(
+ False,
+ False,
+ "airflow.providers.standard.triggers.temporal.DateTimeTrigger",
+ id="active",
+ ),
+ pytest.param(False, True, None, id="stale"),
+ pytest.param(True, False, None, id="paused"),
+ pytest.param(True, False, None, id="stale-paused"),
+ ],
+ )
+ def test_delete_unreferenced_triggers(self, dag_maker, session, paused,
stale, expected_classpath):
+ self.job_runner = SchedulerJobRunner(job=Job())
+
+ classpath, kwargs = DateTimeTrigger(timezone.utcnow()).serialize()
+ asset1 = Asset(
+ name="test_asset_1",
+ watchers=[AssetWatcher(name="test", trigger={"classpath":
classpath, "kwargs": kwargs})],
+ )
+ with dag_maker(dag_id="dag", schedule=[asset1], session=session) as
dag:
+ EmptyOperator(task_id="task")
+ dags = {"dag": dag}
+
+ def _update_references() -> None:
+ asset_op = AssetModelOperation.collect(dags)
+ orm_assets = asset_op.sync_assets(session=session)
+ session.flush()
+ asset_op.add_dag_asset_references(orm_dags, orm_assets,
session=session)
+ asset_op.activate_assets_if_possible(orm_assets.values(),
session=session)
+ asset_op.add_asset_trigger_references(orm_assets, session=session)
+ session.flush()
+
+ # Initial setup.
+ orm_dags = DagModelOperation({"dag": dag}, "testing",
None).add_dags(session=session)
+ _update_references()
+ assert session.scalars(select(Trigger.classpath)).one() == classpath
+
+ # Simulate dag state change.
+ orm_dags["dag"].is_paused = paused
+ orm_dags["dag"].is_stale = stale
+ _update_references()
+ assert session.scalars(select(Trigger.classpath)).one() == classpath
+
+ # Unreferenced trigger should be removed.
+ self.job_runner._remove_unreferenced_triggers(session=session)
+ assert session.scalars(select(Trigger.classpath)).one_or_none() ==
expected_classpath
+
def test_misconfigured_dags_doesnt_crash_scheduler(self, session,
dag_maker, caplog):
"""Test that if dagrun creation throws an exception, the scheduler
doesn't crash"""
with dag_maker("testdag1", serialized=True):
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index f0946f91ebd..2ffa92b1997 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -43,7 +43,6 @@ from airflow.jobs.triggerer_job_runner import (
messages,
)
from airflow.models import DagBag, DagModel, DagRun, TaskInstance, Trigger
-from airflow.models.baseoperator import BaseOperator
from airflow.models.connection import Connection
from airflow.models.dag import DAG
from airflow.models.dag_version import DagVersion
@@ -54,7 +53,7 @@ from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.triggers.temporal import DateTimeTrigger,
TimeDeltaTrigger
-from airflow.sdk import BaseHook
+from airflow.sdk import BaseHook, BaseOperator
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.triggers.testing import FailureTrigger, SuccessTrigger
from airflow.utils.state import State, TaskInstanceState
