Replace external_trigger check with DagRunType

airflow/api/client/local_client.py

@@ -54,7 +54,6 @@ def trigger_dag(
                 "data_interval_start": dag_run.data_interval_start,
                 "data_interval_end": dag_run.data_interval_end,
                 "end_date": dag_run.end_date,
-                "external_trigger": dag_run.external_trigger,
                 "last_scheduling_decision": dag_run.last_scheduling_decision,
                 "logical_date": dag_run.logical_date,
                 "run_type": dag_run.run_type,

airflow/api/common/mark_tasks.py

@@ -145,7 +145,7 @@ def get_run_ids(dag: DAG, run_id: str, future: bool, past: bool, session: SASess
     if current_dagrun.logical_date is None:
         return [run_id]
 
-    last_dagrun = dag.get_last_dagrun(include_externally_triggered=True, session=session)
+    last_dagrun = dag.get_last_dagrun(include_manually_triggered=True, session=session)
     first_dagrun = session.scalar(
         select(DagRun)
         .where(DagRun.dag_id == dag.dag_id, DagRun.logical_date.is_not(None))

airflow/api/common/trigger_dag.py

@@ -113,7 +113,6 @@ def _trigger_dag(
         conf=run_conf,
         run_type=DagRunType.MANUAL,
         triggered_by=triggered_by,
-        external_trigger=True,
         dag_version=dag_version,
         state=DagRunState.QUEUED,
         session=session,

airflow/api_connexion/endpoints/dag_run_endpoint.py

@@ -185,7 +185,6 @@ def _fetch_dag_runs(
         "start_date",
         "end_date",
         "updated_at",
-        "external_trigger",
         "conf",
     ]
     query = apply_sorting(query, order_by, to_replace, allowed_sort_attrs)
@@ -365,7 +364,6 @@ def post_dag_run(*, dag_id: str, session: Session = NEW_SESSION) -> APIResponse:
                 conf=post_body.get("conf"),
                 run_type=DagRunType.MANUAL,
                 triggered_by=DagRunTriggeredByType.REST_API,
-                external_trigger=True,
                 dag_version=DagVersion.get_latest_version(dag.dag_id),
                 state=DagRunState.QUEUED,
                 session=session,

airflow/api_connexion/schemas/dag_run_schema.py

@@ -68,7 +68,6 @@ class Meta:
     start_date = auto_field(dump_only=True)
     end_date = auto_field(dump_only=True)
     state = DagStateField(dump_only=True)
-    external_trigger = auto_field(dump_default=True, dump_only=True)
     conf = ConfObject()
     data_interval_start = auto_field(validate=validate_istimezone)
     data_interval_end = auto_field(validate=validate_istimezone)

airflow/api_fastapi/core_api/datamodels/dag_run.py

@@ -71,7 +71,6 @@ class DAGRunResponse(BaseModel):
     last_scheduling_decision: datetime | None
     run_type: DagRunType
     state: DagRunState
-    external_trigger: bool
     triggered_by: DagRunTriggeredByType
     conf: dict
     note: str | None

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -8527,9 +8527,6 @@ components:
           $ref: '#/components/schemas/DagRunType'
         state:
           $ref: '#/components/schemas/DagRunState'
-        external_trigger:
-          type: boolean
-          title: External Trigger
         triggered_by:
           $ref: '#/components/schemas/DagRunTriggeredByType'
         conf:
@@ -8554,7 +8551,6 @@ components:
       - last_scheduling_decision
       - run_type
       - state
-      - external_trigger
       - triggered_by
       - conf
       - note

airflow/api_fastapi/core_api/routes/public/assets.py

@@ -297,7 +297,6 @@ def materialize_asset(
         run_after=run_after,
         run_type=DagRunType.MANUAL,
         triggered_by=DagRunTriggeredByType.REST_API,
-        external_trigger=True,
         dag_version=DagVersion.get_latest_version(dag_id, session=session),
         state=DagRunState.QUEUED,
         session=session,

airflow/api_fastapi/core_api/routes/public/dag_run.py

@@ -286,7 +286,6 @@ def get_dag_runs(
                     "start_date",
                     "end_date",
                     "updated_at",
-                    "external_trigger",
                     "conf",
                 ],
                 DagRun,
@@ -367,7 +366,6 @@ def trigger_dag_run(
             conf=params["conf"],
             run_type=DagRunType.MANUAL,
             triggered_by=DagRunTriggeredByType.REST_API,
-            external_trigger=True,
             dag_version=DagVersion.get_latest_version(dag.dag_id),
             state=DagRunState.QUEUED,
             session=session,
@@ -421,7 +419,6 @@ def get_list_dag_runs_batch(
             "start_date",
             "end_date",
             "updated_at",
-            "external_trigger",
             "conf",
         ],
         DagRun,

airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -230,7 +230,6 @@ class DagRun(StrictBaseModel):
     clear_number: int
     run_type: DagRunType
     conf: Annotated[dict[str, Any], Field(default_factory=dict)]
-    external_trigger: bool = False
 
 
 class TIRunContext(BaseModel):

airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -160,7 +160,6 @@ def ti_run(
                 DR.run_type,
                 DR.conf,
                 DR.logical_date,
-                DR.external_trigger,
             ).filter_by(dag_id=dag_id, run_id=run_id)
         ).one_or_none()
 

airflow/cli/commands/remote_commands/task_command.py

@@ -139,7 +139,6 @@ def _get_dag_run(
             dag_id=dag.dag_id,
             run_id=logical_date_or_run_id,
             run_type=DagRunType.MANUAL,
-            external_trigger=True,
             logical_date=dag_run_logical_date,
             data_interval=data_interval,
             run_after=run_after,

airflow/example_dags/plugins/event_listener.py

@@ -133,9 +133,9 @@ def on_dag_run_failed(dag_run: DagRun, msg: str):
     print("Dag run  in failure state")
     dag_id = dag_run.dag_id
     run_id = dag_run.run_id
-    external_trigger = dag_run.external_trigger
+    run_type = dag_run.run_type
 
-    print(f"Dag information:{dag_id} Run id: {run_id} external trigger: {external_trigger}")
+    print(f"Dag information:{dag_id} Run id: {run_id} Run type: {run_type}")
     print(f"Failed with message: {msg}")
 
 

airflow/exceptions.py

@@ -270,7 +270,6 @@ def serialize(self):
             state=self.dag_run.state,
             dag_id=self.dag_run.dag_id,
             run_id=self.dag_run.run_id,
-            external_trigger=self.dag_run.external_trigger,
             run_type=self.dag_run.run_type,
         )
         dag_run.id = self.dag_run.id

airflow/jobs/scheduler_job_runner.py

@@ -1459,8 +1459,8 @@ def _update_state(dag: DAG, dag_run: DagRun):
             dag_run.start_date = timezone.utcnow()
             if (
                 dag.timetable.periodic
+                and dag_run.run_type != DagRunType.MANUAL
                 and dag_run.triggered_by != DagRunTriggeredByType.ASSET
-                and not dag_run.external_trigger
                 and dag_run.clear_number < 1
             ):
                 # TODO: Logically, this should be DagRunInfo.run_after, but the
@@ -1469,7 +1469,7 @@ def _update_state(dag: DAG, dag_run: DagRun):
                 # a field on DagRun for this instead of relying on the run
                 # always happening immediately after the data interval.
                 # We only publish these metrics for scheduled dag runs and only
-                # when ``external_trigger`` is *False* and ``clear_number`` is 0.
+                # when ``run_type`` is *MANUAL* and ``clear_number`` is 0.
                 expected_start_date = dag.get_run_data_interval(dag_run).end
                 schedule_delay = dag_run.start_date - expected_start_date
                 # Publish metrics twice with backward compatible name, and then with tags

airflow/migrations/versions/0059_3_0_0_remove_external_trigger_field.py

@@ -0,0 +1,58 @@
+#
+# 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.
+
+"""
+remove external_trigger field.
+
+Revision ID: e00344393f31
+Revises: 6a9e7a527a88
+Create Date: 2025-02-04 16:45:49.068781
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "e00344393f31"
+down_revision = "6a9e7a527a88"
+branch_labels = None
+depends_on = None
+airflow_version = "3.0.0"
+
+
+def upgrade():
+    """Apply remove external_trigger field."""
+    with op.batch_alter_table("dag_run", schema=None) as batch_op:
+        batch_op.drop_column("external_trigger")
+
+
+def downgrade():
+    """Unapply remove external_trigger field."""
+    with op.batch_alter_table("dag_run", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("external_trigger", sa.BOOLEAN(), autoincrement=False, nullable=True))
+    # restore external_trigger field based on run_type
+    dag_run_table = sa.table(
+        "dag_run", sa.column("external_trigger", sa.BOOLEAN()), sa.column("run_type", sa.String())
+    )
+    op.execute(
+        dag_run_table.update().values(
+            external_trigger=sa.case([(dag_run_table.c.run_type == "manual", True)], else_=False)
+        )
+    )

airflow/models/dag.py

@@ -180,7 +180,7 @@ def _get_model_data_interval(
     return DataInterval(start, end)
 
 
-def get_last_dagrun(dag_id, session, include_externally_triggered=False):
+def get_last_dagrun(dag_id, session, include_manually_triggered=False):
     """
     Return the last dag run for a dag, None if there was none.
 
@@ -189,8 +189,8 @@ def get_last_dagrun(dag_id, session, include_externally_triggered=False):
     """
     DR = DagRun
     query = select(DR).where(DR.dag_id == dag_id, DR.logical_date.is_not(None))
-    if not include_externally_triggered:
-        query = query.where(DR.external_trigger == expression.false())
+    if not include_manually_triggered:
+        query = query.where(DR.run_type != DagRunType.MANUAL)
     query = query.order_by(DR.logical_date.desc())
     return session.scalar(query.limit(1))
 
@@ -252,7 +252,6 @@ def _create_orm_dagrun(
     data_interval: DataInterval | None,
     run_after: datetime,
     start_date: datetime | None,
-    external_trigger: bool,
     conf: Any,
     state: DagRunState | None,
     run_type: DagRunType,
@@ -268,7 +267,6 @@ def _create_orm_dagrun(
         logical_date=logical_date,
         start_date=start_date,
         run_after=run_after,
-        external_trigger=external_trigger,
         conf=conf,
         state=state,
         run_type=run_type,
@@ -710,16 +708,16 @@ def iter_dagrun_infos_between(
                 break
 
     @provide_session
-    def get_last_dagrun(self, session=NEW_SESSION, include_externally_triggered=False):
+    def get_last_dagrun(self, session=NEW_SESSION, include_manually_triggered=False):
         return get_last_dagrun(
-            self.dag_id, session=session, include_externally_triggered=include_externally_triggered
+            self.dag_id, session=session, include_manually_triggered=include_manually_triggered
         )
 
     @provide_session
-    def has_dag_runs(self, session=NEW_SESSION, include_externally_triggered=True) -> bool:
+    def has_dag_runs(self, session=NEW_SESSION, include_manually_triggered=True) -> bool:
         return (
             get_last_dagrun(
-                self.dag_id, session=session, include_externally_triggered=include_externally_triggered
+                self.dag_id, session=session, include_manually_triggered=include_manually_triggered
             )
             is not None
         )
@@ -1767,7 +1765,6 @@ def create_dagrun(
         conf: dict | None = None,
         run_type: DagRunType,
         triggered_by: DagRunTriggeredByType,
-        external_trigger: bool = False,
         dag_version: DagVersion | None = None,
         state: DagRunState,
         start_date: datetime | None = None,
@@ -1839,7 +1836,6 @@ def create_dagrun(
             data_interval=data_interval,
             run_after=timezone.coerce_datetime(run_after),
             start_date=timezone.coerce_datetime(start_date),
-            external_trigger=external_trigger,
             conf=conf,
             state=state,
             run_type=run_type,
@@ -2206,9 +2202,9 @@ def get_current(cls, dag_id: str, session=NEW_SESSION) -> DagModel:
         return session.scalar(select(cls).where(cls.dag_id == dag_id))
 
     @provide_session
-    def get_last_dagrun(self, session=NEW_SESSION, include_externally_triggered=False):
+    def get_last_dagrun(self, session=NEW_SESSION, include_manually_triggered=False):
         return get_last_dagrun(
-            self.dag_id, session=session, include_externally_triggered=include_externally_triggered
+            self.dag_id, session=session, include_manually_triggered=include_manually_triggered
         )
 
     def get_is_paused(self, *, session: Session | None = None) -> bool: