Also expose scheduled_dttm in the Gantt Chart

Author: smyjt

airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/gantt.py

@@ -30,6 +30,7 @@ class GanttTaskInstance(BaseModel):
     task_display_name: str
     try_number: int
     state: TaskInstanceState | None
+    scheduled_dttm: datetime | None
     queued_dttm: datetime | None
     start_date: datetime | None
     end_date: datetime | None

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -2237,6 +2237,12 @@ components:
           anyOf:
           - $ref: '#/components/schemas/TaskInstanceState'
           - type: 'null'
+        scheduled_dttm:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Scheduled Dttm
         queued_dttm:
           anyOf:
           - type: string
@@ -2269,6 +2275,7 @@ components:
       - task_display_name
       - try_number
       - state
+      - scheduled_dttm
       - queued_dttm
       - start_date
       - end_date

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/gantt.py

@@ -67,6 +67,7 @@ def get_gantt_data(
         TaskInstance.task_display_name.label("task_display_name"),  # type: ignore[attr-defined]
         TaskInstance.try_number.label("try_number"),
         TaskInstance.state.label("state"),
+        TaskInstance.scheduled_dttm.label("scheduled_dttm"),
         TaskInstance.queued_dttm.label("queued_dttm"),
         TaskInstance.start_date.label("start_date"),
         TaskInstance.end_date.label("end_date"),
@@ -82,6 +83,7 @@ def get_gantt_data(
         TaskInstanceHistory.task_display_name.label("task_display_name"),
         TaskInstanceHistory.try_number.label("try_number"),
         TaskInstanceHistory.state.label("state"),
+        TaskInstanceHistory.scheduled_dttm.label("scheduled_dttm"),
         TaskInstanceHistory.queued_dttm.label("queued_dttm"),
         TaskInstanceHistory.start_date.label("start_date"),
         TaskInstanceHistory.end_date.label("end_date"),
@@ -108,6 +110,7 @@ def get_gantt_data(
             task_display_name=row.task_display_name,
             try_number=row.try_number,
             state=row.state,
+            scheduled_dttm=row.scheduled_dttm,
             queued_dttm=row.queued_dttm,
             start_date=row.start_date,
             end_date=row.end_date,

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -8046,6 +8046,18 @@ export const $GanttTaskInstance = {
                 }
             ]
         },
+        scheduled_dttm: {
+            anyOf: [
+                {
+                    type: 'string',
+                    format: 'date-time'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Scheduled Dttm'
+        },
         queued_dttm: {
             anyOf: [
                 {
@@ -8094,7 +8106,7 @@ export const $GanttTaskInstance = {
         }
     },
     type: 'object',
-    required: ['task_id', 'task_display_name', 'try_number', 'state', 'queued_dttm', 'start_date', 'end_date'],
+    required: ['task_id', 'task_display_name', 'try_number', 'state', 'scheduled_dttm', 'queued_dttm', 'start_date', 'end_date'],
     title: 'GanttTaskInstance',
     description: 'Task instance data for Gantt chart.'
 } as const;

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -1983,6 +1983,7 @@ export type GanttTaskInstance = {
     task_display_name: string;
     try_number: number;
     state: TaskInstanceState | null;
+    scheduled_dttm: string | null;
     queued_dttm: string | null;
     start_date: string | null;
     end_date: string | null;

airflow-core/src/airflow/ui/src/layouts/Details/Gantt/utils.ts

@@ -39,6 +39,7 @@ export type GanttDataItem = {
   isGroup?: boolean | null;
   isMapped?: boolean | null;
   isQueued?: boolean;
+  isScheduled?: boolean;
   state?: TaskInstanceState | null;
   taskId: string;
   tryNumber?: number;
@@ -128,6 +129,22 @@ export const transformGanttData = ({
               const endTime = hasTaskRunning ? dayjs().toISOString() : tryInstance.end_date;
               const items: Array<GanttDataItem> = [];
 
+              // Scheduled segment: from scheduled_dttm to queued_dttm (or start_date if no queued_dttm)
+              if (tryInstance.scheduled_dttm !== null) {
+                const scheduledEnd = tryInstance.queued_dttm ?? tryInstance.start_date;
+
+                items.push({
+                  isGroup: false,
+                  isMapped: tryInstance.is_mapped,
+                  isScheduled: true,
+                  state: "scheduled" as TaskInstanceState,
+                  taskId: tryInstance.task_id,
+                  tryNumber: tryInstance.try_number,
+                  x: [dayjs(tryInstance.scheduled_dttm).toISOString(), dayjs(scheduledEnd).toISOString()],
+                  y: tryInstance.task_display_name,
+                });
+              }
+
               // Queue segment: from queued_dttm to start_date
               if (tryInstance.queued_dttm !== null) {
                 items.push({
@@ -351,6 +368,10 @@ export const createChartOptions = ({
           label(tooltipItem: TooltipItem<"bar">) {
             const taskInstance = data[tooltipItem.dataIndex];
 
+            if (taskInstance?.isScheduled) {
+              return `${translate("state")}: ${translate("states.scheduled")}`;
+            }
+
             if (taskInstance?.isQueued) {
               return `${translate("state")}: ${translate("states.queued")}`;
             }

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_gantt.py

@@ -51,6 +51,7 @@
     "task_display_name": TASK_DISPLAY_NAME,
     "try_number": 1,
     "state": "success",
+    "scheduled_dttm": "2024-11-30T09:50:00Z",
     "queued_dttm": "2024-11-30T09:55:00Z",
     "start_date": "2024-11-30T10:00:00Z",
     "end_date": "2024-11-30T10:05:00Z",
@@ -63,6 +64,7 @@
     "task_display_name": TASK_DISPLAY_NAME_2,
     "try_number": 1,
     "state": "failed",
+    "scheduled_dttm": "2024-11-30T10:02:00Z",
     "queued_dttm": "2024-11-30T10:03:00Z",
     "start_date": "2024-11-30T10:05:00Z",
     "end_date": "2024-11-30T10:10:00Z",
@@ -75,6 +77,7 @@
     "task_display_name": TASK_DISPLAY_NAME_3,
     "try_number": 1,
     "state": "running",
+    "scheduled_dttm": None,
     "queued_dttm": None,
     "start_date": "2024-11-30T10:10:00Z",
     "end_date": None,
@@ -119,18 +122,21 @@ def setup(dag_maker, session=None):
         if ti.task_id == TASK_ID:
             ti.state = TaskInstanceState.SUCCESS
             ti.try_number = 1
+            ti.scheduled_dttm = pendulum.DateTime(2024, 11, 30, 9, 50, 0, tzinfo=pendulum.UTC)
             ti.queued_dttm = pendulum.DateTime(2024, 11, 30, 9, 55, 0, tzinfo=pendulum.UTC)
             ti.start_date = pendulum.DateTime(2024, 11, 30, 10, 0, 0, tzinfo=pendulum.UTC)
             ti.end_date = pendulum.DateTime(2024, 11, 30, 10, 5, 0, tzinfo=pendulum.UTC)
         elif ti.task_id == TASK_ID_2:
             ti.state = TaskInstanceState.FAILED
             ti.try_number = 1
+            ti.scheduled_dttm = pendulum.DateTime(2024, 11, 30, 10, 2, 0, tzinfo=pendulum.UTC)
             ti.queued_dttm = pendulum.DateTime(2024, 11, 30, 10, 3, 0, tzinfo=pendulum.UTC)
             ti.start_date = pendulum.DateTime(2024, 11, 30, 10, 5, 0, tzinfo=pendulum.UTC)
             ti.end_date = pendulum.DateTime(2024, 11, 30, 10, 10, 0, tzinfo=pendulum.UTC)
         elif ti.task_id == TASK_ID_3:
             ti.state = TaskInstanceState.RUNNING
             ti.try_number = 1
+            ti.scheduled_dttm = None
             ti.queued_dttm = None
             ti.start_date = pendulum.DateTime(2024, 11, 30, 10, 10, 0, tzinfo=pendulum.UTC)
             ti.end_date = None
@@ -312,13 +318,16 @@ def test_sorted_by_task_id_and_try_number(self, test_client):
         sorted_tis = sorted(task_instances, key=lambda x: (x["task_id"], x["try_number"]))
         assert task_instances == sorted_tis
 
-    def test_queued_dttm_is_returned(self, test_client):
+    def test_timing_fields_are_returned(self, test_client):
         response = test_client.get(f"/gantt/{DAG_ID}/run_1")
         assert response.status_code == 200
         data = response.json()
         tis = {ti["task_id"]: ti for ti in data["task_instances"]}
+        assert tis[TASK_ID]["scheduled_dttm"] == "2024-11-30T09:50:00Z"
         assert tis[TASK_ID]["queued_dttm"] == "2024-11-30T09:55:00Z"
+        assert tis[TASK_ID_2]["scheduled_dttm"] == "2024-11-30T10:02:00Z"
         assert tis[TASK_ID_2]["queued_dttm"] == "2024-11-30T10:03:00Z"
+        assert tis[TASK_ID_3]["scheduled_dttm"] is None
         assert tis[TASK_ID_3]["queued_dttm"] is None
 
     def test_should_response_401(self, unauthenticated_test_client):