Add team_name column to trigger table for multi-team triggerer support

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``a1b2c3d4e5f6`` (head) | ``a7f3b2c1d4e5`` | ``3.3.0``         | Add version_data to dag_version.                             |
+| ``acc215baed80`` (head) | ``a1b2c3d4e5f6`` | ``3.3.0``         | Add team_name to trigger table.                              |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``a1b2c3d4e5f6``        | ``a7f3b2c1d4e5`` | ``3.3.0``         | Add version_data to dag_version.                             |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``a7f3b2c1d4e5``        | ``b8f3e4a1d2c9`` | ``3.3.0``         | Add allow_producer_teams column to                           |
 |                         |                  |                   | dag_schedule_asset_reference table.                          |

airflow-core/src/airflow/migrations/versions/0116_3_3_0_add_team_name_to_trigger_table.py

@@ -0,0 +1,55 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add team_name to trigger table.
+
+Revision ID: acc215baed80
+Revises: a1b2c3d4e5f6
+Create Date: 2026-05-21 21:38:00.122692
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "acc215baed80"
+down_revision = "a1b2c3d4e5f6"
+branch_labels = None
+depends_on = None
+airflow_version = "3.3.0"
+
+
+def upgrade():
+    """Add team_name to trigger table."""
+    with op.batch_alter_table("trigger", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("team_name", sa.String(length=50), nullable=True))
+        batch_op.create_index(batch_op.f("idx_trigger_team_name"), ["team_name"], unique=False)
+        batch_op.create_foreign_key(
+            batch_op.f("trigger_team_name_fkey"), "team", ["team_name"], ["name"], ondelete="SET NULL"
+        )
+
+
+def downgrade():
+    """Remove team_name from trigger table."""
+    with op.batch_alter_table("trigger", schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f("trigger_team_name_fkey"), type_="foreignkey")
+        batch_op.drop_index(batch_op.f("idx_trigger_team_name"))
+        batch_op.drop_column("team_name")

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

@@ -24,7 +24,7 @@
 from traceback import format_exception
 from typing import TYPE_CHECKING, Any
 
-from sqlalchemy import Integer, String, Text, delete, func, or_, select, update
+from sqlalchemy import ForeignKey, Integer, String, Text, delete, func, or_, select, update
 from sqlalchemy.ext.associationproxy import association_proxy
 from sqlalchemy.orm import Mapped, Session, mapped_column, relationship, selectinload
 from sqlalchemy.sql.functions import coalesce
@@ -100,6 +100,15 @@ class Trigger(Base):
     triggerer_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
     queue: Mapped[str | None] = mapped_column(String(256), nullable=True)
 
+    # Denormalized from dag_bundle_team to keep the triggerer's ~1s polling queries join-free,
+    # especially since it's eventually consistent and trigger rows are ephemeral.
+    # Without this, filtering by team requires 2-3 joins depending on trigger type.
+    # Performance testing confirmed the denormalized column avoids measurable overhead in the
+    # triggerer loop under load.
+    team_name: Mapped[str | None] = mapped_column(
+        String(50), ForeignKey("team.name", ondelete="SET NULL"), nullable=True, index=True
+    )
+
     triggerer_job = relationship(
         "Job",
         primaryjoin="Job.id == Trigger.triggerer_id",
@@ -122,12 +131,14 @@ def __init__(
         kwargs: dict[str, Any],
         created_date: datetime.datetime | None = None,
         queue: str | None = None,
+        team_name: str | None = None,
     ) -> None:
         super().__init__()
         self.classpath = classpath
         self.encrypted_kwargs = self.encrypt_kwargs(kwargs)
         self.created_date = created_date or timezone.utcnow()
         self.queue = queue
+        self.team_name = team_name
 
     @property
     def kwargs(self) -> dict[str, Any]:

airflow-core/src/airflow/utils/db.py

@@ -116,7 +116,7 @@ class MappedClassProtocol(Protocol):
     "3.1.0": "cc92b33c6709",
     "3.1.8": "509b94a1042d",
     "3.2.0": "1d6611b6ab7c",
-    "3.3.0": "a1b2c3d4e5f6",
+    "3.3.0": "acc215baed80",
 }
 
 # Prefix used to identify tables holding data moved during migration.

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

@@ -83,6 +83,17 @@ def clear_db(session):
     session.commit()
 
 
+def test_trigger_team_name_stored(session, testing_team):
+    trigger = Trigger(
+        classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}, team_name=testing_team.name
+    )
+    session.add(trigger)
+    session.flush()
+
+    loaded = session.get(Trigger, trigger.id)
+    assert loaded.team_name == "testing"
+
+
 def test_fetch_trigger_ids_with_non_task_associations(session):
     # Create triggers
     asset_trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger1", kwargs={})