This is an automated email from the ASF dual-hosted git repository.
jscheffl 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 8546c91a107 select unassigned asset triggers from db (#65792)
8546c91a107 is described below
commit 8546c91a107831b55e9f2543120d9609abf96dae
Author: renat-sagut <[email protected]>
AuthorDate: Tue Apr 28 20:48:16 2026 +0200
select unassigned asset triggers from db (#65792)
* select unassigned triggers from db
* tests
* cleanup
---
airflow-core/src/airflow/models/trigger.py | 7 +-
airflow-core/tests/unit/models/test_trigger.py | 99 ++++++++++++++++++++++++++
2 files changed, 105 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/models/trigger.py
b/airflow-core/src/airflow/models/trigger.py
index 589725ced2c..2949262532e 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -427,7 +427,12 @@ class Trigger(Base):
.where(or_(cls.triggerer_id.is_(None),
cls.triggerer_id.not_in(alive_triggerer_ids)))
.order_by(coalesce(TaskInstance.priority_weight, 0).desc(),
cls.created_date),
# Asset triggers
- select(cls.id).where(cls.assets.any()).order_by(cls.created_date),
+ select(cls.id)
+ .where(
+ cls.assets.any(),
+ or_(cls.triggerer_id.is_(None),
cls.triggerer_id.not_in(alive_triggerer_ids)),
+ )
+ .order_by(cls.created_date),
]
# Process each query while avoiding unnecessary queries when capacity
is reached
diff --git a/airflow-core/tests/unit/models/test_trigger.py
b/airflow-core/tests/unit/models/test_trigger.py
index dfd0f2e99cd..c79a021f8b8 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -919,3 +919,102 @@ def test_kwargs_not_encrypted():
assert trigger.kwargs["param1"] == "value1"
assert trigger.kwargs["param2"] == "value2"
+
+
+def test_asset_trigger_unassigned_included(session):
+ """Asset triggers with triggerer_id=None are returned."""
+ asset = AssetModel("test_asset")
+ trigger = Trigger(classpath="some.trigger", kwargs={})
+ session.add(trigger)
+ session.flush() # ensure trigger.id is available
+ asset.add_trigger(trigger, "test_watcher")
+ session.add(asset)
+ session.flush()
+
+ alive_ids = [100, 200]
+ result = Trigger.get_sorted_triggers(
+ capacity=10,
+ alive_triggerer_ids=alive_ids,
+ queues=None,
+ session=session,
+ )
+ ids = [row[0] for row in result]
+
+ assert trigger.id in ids
+
+
+def test_asset_trigger_dead_triggerer_included(session):
+ """Asset triggers assigned to a dead triggerer are returned."""
+ asset = AssetModel("test_asset")
+ trigger = Trigger(classpath="some.trigger", kwargs={})
+ trigger.triggerer_id = 999 # dead
+ session.add(trigger)
+ session.flush()
+ asset.add_trigger(trigger, "test_watcher")
+ session.add(asset)
+ session.flush()
+
+ alive_ids = [100, 200]
+ result = Trigger.get_sorted_triggers(
+ capacity=10,
+ alive_triggerer_ids=alive_ids,
+ queues=None,
+ session=session,
+ )
+ ids = [row[0] for row in result]
+
+ assert trigger.id in ids
+
+
+def test_asset_trigger_alive_triggerer_excluded(session):
+ """Asset triggers assigned to a living triggerer are not returned."""
+ asset = AssetModel("test_asset")
+ trigger = Trigger(classpath="some.trigger", kwargs={})
+ trigger.triggerer_id = 100 # alive
+ session.add(trigger)
+ session.flush()
+ asset.add_trigger(trigger, "test_watcher")
+ session.add(asset)
+ session.flush()
+
+ alive_ids = [100, 200]
+ result = Trigger.get_sorted_triggers(
+ capacity=10,
+ alive_triggerer_ids=alive_ids,
+ queues=None,
+ session=session,
+ )
+ ids = [row[0] for row in result]
+
+ assert trigger.id not in ids
+
+
+def test_asset_trigger_ordering_and_capacity(session):
+ """Asset triggers are ordered by created_date (oldest first) and respect
capacity."""
+ now = datetime.datetime(2025, 1, 1, tzinfo=timezone.utc)
+ asset = AssetModel("test_asset")
+ triggers = []
+ for i in range(5):
+ trigger = Trigger(
+ classpath="some.trigger",
+ kwargs={},
+ created_date=now + datetime.timedelta(hours=i),
+ )
+ trigger.triggerer_id = None # all unassigned
+ session.add(trigger)
+ session.flush()
+ asset.add_trigger(trigger, f"watcher_{i}")
+ triggers.append(trigger)
+ session.add(asset)
+ session.flush()
+
+ result = Trigger.get_sorted_triggers(
+ capacity=3,
+ alive_triggerer_ids=[],
+ queues=None,
+ session=session,
+ )
+ ids = [row[0] for row in result]
+
+ # Only the three oldest should be returned, in order
+ assert ids == [triggers[0].id, triggers[1].id, triggers[2].id]
