Replace external_trigger check with DagRunType (apache#45961)

* Replace external_trigger check with DagRunType

- Fix test_handle_multiple_columns_unique_constraint_error

- Resolve ashb code review

- Add newsfragment

- Replace external_trigger is False logic with run_type is SCHEDULED

- Fix _emit_true_scheduling_delay_stats_for_finished_state

- Fix migrations fix after rebasing to latest main

Fix test_exceptions and static check after rebasing

Fix test_dag_run and erd static check

* fixup! Fix run_type logic, remove unused externally_triggered_type

* fixup! Restore external_trigger value when downgrade

* Remove externally_trigger reference in frontend

The 'triggered_by' field used to only be shown when the run is
externally triggered. Rather than using run_type to maintain the
conditional, I decided to render the field unconditionally instead,
since the value could be useful for scheduler-created runs too (for
example, distinguishing whether a run is asset-triggered or
time-scheduled).

* docs(newsfragments): add migration rules

---------

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>
Co-authored-by: Wei Lee <weilee.rx@gmail.com>

Author: 3 people

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):