Fix N+1 queries in trigger asset event submission

airflow-core/src/airflow/models/trigger.py

@@ -286,7 +286,11 @@ def submit_event(cls, trigger_id, event: TriggerEvent, *, session: Session = NEW
             handle_event_submit(event, task_instance=task_instance, session=session)
 
         # Send an event to assets
-        trigger = session.scalars(select(cls).where(cls.id == trigger_id)).one_or_none()
+        trigger = session.scalars(
+            select(cls)
+            .where(cls.id == trigger_id)
+            .options(selectinload(cls.asset_watchers).selectinload(AssetWatcherModel.asset))
+        ).one_or_none()
         if trigger is None:
             # Already deleted for some reason
             return

airflow-core/tests/unit/models/test_trigger.py

@@ -50,6 +50,7 @@
 from airflow.utils.session import create_session
 from airflow.utils.state import State
 
+from tests_common.test_utils.asserts import count_queries
 from tests_common.test_utils.config import conf_vars
 
 if TYPE_CHECKING:
@@ -234,6 +235,38 @@ def test_submit_event(mock_callback_handle_event, session, create_task_instance)
     mock_callback_handle_event.assert_called_once_with(event, session)
 
 
+@patch("airflow.models.trigger.AssetManager.register_asset_change")
+def test_submit_event_no_n_plus_one_for_assets(_, session):
+    """Ensure asset notifications do not trigger per-asset lazy-load queries."""
+
+    def _create_trigger_with_assets(asset_count: int) -> int:
+        trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={})
+        session.add(trigger)
+        session.flush()
+
+        for i in range(asset_count):
+            asset = AssetModel(name=f"asset_{asset_count}_{i}")
+            asset.add_trigger(trigger, f"watcher_{i}")
+            session.add(asset)
+
+        session.commit()
+        session.expire_all()
+        return trigger.id
+
+    def _submit_event_query_count(trigger_id: int) -> int:
+        with count_queries(session=session) as query_result:
+            Trigger.submit_event(trigger_id, TriggerEvent("payload"), session=session)
+        return sum(query_result.values())
+
+    small_query_count = _submit_event_query_count(_create_trigger_with_assets(1))
+    larger_query_count = _submit_event_query_count(_create_trigger_with_assets(5))
+
+    assert larger_query_count - small_query_count < 3, (
+        f"Added 4 assets but query count increased by {larger_query_count - small_query_count} "
+        f"({small_query_count} -> {larger_query_count}), suggesting n+1 queries for trigger assets"
+    )
+
+
 def test_submit_failure(session, create_task_instance):
     """
     Tests that failures submitted to a trigger fail their dependent