Merge branch 'main' into add-workers-celery-k8s-container-lifecycle-hooks

Author: Miretpl

.github/CODEOWNERS

@@ -149,3 +149,6 @@ Dockerfile.ci @potiuk @ashb @gopidesupavan @amoghrajesh @jscheffl @bugraoz93 @ka
 
 # Shared Libraries
 /shared/ @ashb @amoghrajesh @potiuk
+
+# RMs on release documents
+/dev/README_RELEASE_*.md @potiuk @jscheffl @vincbeck @shahar1 @jedcunningham @bugraoz93

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

@@ -24,6 +24,7 @@
 
 from airflow._shared.timezones import timezone
 from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse
 from airflow.utils.state import DagRunState
 from airflow.utils.types import DagRunType
 
@@ -79,6 +80,7 @@ class GridRunsResponse(BaseModel):
     run_after: datetime
     state: DagRunState | None
     run_type: DagRunType
+    dag_versions: list[DagVersionResponse] = []
     has_missed_deadline: bool
 
     @computed_field

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

@@ -32,6 +32,7 @@ class LightGridTaskInstanceSummary(BaseModel):
     child_states: dict[TaskInstanceState | None, int] | None
     min_start_date: datetime | None
     max_end_date: datetime | None
+    dag_version_number: int | None = None
 
 
 class GridTISummaries(BaseModel):

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

@@ -2305,6 +2305,12 @@ components:
           - type: 'null'
         run_type:
           $ref: '#/components/schemas/DagRunType'
+        dag_versions:
+          items:
+            $ref: '#/components/schemas/DagVersionResponse'
+          type: array
+          title: Dag Versions
+          default: []
         has_missed_deadline:
           type: boolean
           title: Has Missed Deadline
@@ -2575,6 +2581,11 @@ components:
             format: date-time
           - type: 'null'
           title: Max End Date
+        dag_version_number:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Dag Version Number
       type: object
       required:
       - task_id

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

@@ -23,7 +23,7 @@
 import structlog
 from fastapi import Depends, HTTPException, status
 from sqlalchemy import exists, select
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import joinedload, load_only, selectinload
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
@@ -58,11 +58,13 @@
     get_task_group_children_getter,
     task_group_to_dict_grid,
 )
+from airflow.models.dag import DagModel
 from airflow.models.dag_version import DagVersion
 from airflow.models.dagrun import DagRun
 from airflow.models.deadline import Deadline
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.taskinstance import TaskInstance
+from airflow.models.taskinstancehistory import TaskInstanceHistory
 
 log = structlog.get_logger(logger_name=__name__)
 grid_router = AirflowRouter(prefix="/grid", tags=["Grid"])
@@ -282,17 +284,33 @@ def get_grid_runs(
         .correlate(DagRun)
         .label("has_missed_deadline")
     )
-    base_query = select(
-        DagRun.dag_id,
-        DagRun.run_id,
-        DagRun.queued_at,
-        DagRun.start_date,
-        DagRun.end_date,
-        DagRun.run_after,
-        DagRun.state,
-        DagRun.run_type,
-        has_missed_deadline,
-    ).where(DagRun.dag_id == dag_id)
+    base_query = (
+        select(DagRun, has_missed_deadline)
+        .where(DagRun.dag_id == dag_id)
+        .options(
+            load_only(
+                DagRun.dag_id,
+                DagRun.run_id,
+                DagRun.queued_at,
+                DagRun.start_date,
+                DagRun.end_date,
+                DagRun.run_after,
+                DagRun.state,
+                DagRun.run_type,
+                DagRun.bundle_version,
+            ),
+            joinedload(DagRun.dag_model).load_only(DagModel._dag_display_property_value),
+            joinedload(DagRun.created_dag_version).joinedload(DagVersion.bundle),
+            selectinload(DagRun.task_instances)
+            .load_only(TaskInstance.dag_version_id)
+            .joinedload(TaskInstance.dag_version)
+            .joinedload(DagVersion.bundle),
+            selectinload(DagRun.task_instances_histories)
+            .load_only(TaskInstanceHistory.dag_version_id)
+            .joinedload(TaskInstanceHistory.dag_version)
+            .joinedload(DagVersion.bundle),
+        )
+    )
 
     # This comparison is to fall back to DAG timetable when no order_by is provided
     if order_by.value == [order_by.get_primary_key_string()]:
@@ -309,8 +327,14 @@ def get_grid_runs(
         offset=offset,
         filters=[run_after, run_type, state, triggering_user],
         limit=limit,
+        return_total_entries=False,
     )
-    return [GridRunsResponse(**row._mapping) for row in session.execute(dag_runs_select_filter)]
+    results = session.execute(dag_runs_select_filter).unique().all()
+    grid_runs = []
+    for run, has_missed in results:
+        run.has_missed_deadline = has_missed
+        grid_runs.append(GridRunsResponse.model_validate(run, from_attributes=True))
+    return grid_runs
 
 
 @grid_router.get(
@@ -363,7 +387,9 @@ def get_grid_ti_summaries(
                 TaskInstance.dag_version_id,
                 TaskInstance.start_date,
                 TaskInstance.end_date,
+                DagVersion.version_number,
             )
+            .outerjoin(DagVersion, TaskInstance.dag_version_id == DagVersion.id)
             .where(TaskInstance.dag_id == dag_id)
             .where(
                 TaskInstance.run_id == run_id,
@@ -386,6 +412,7 @@ def get_grid_ti_summaries(
                 "state": ti.state,
                 "start_date": ti.start_date,
                 "end_date": ti.end_date,
+                "dag_version_number": ti.version_number,
             }
         )
     serdag = _get_serdag(

airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py

@@ -72,11 +72,18 @@ def _get_aggs_for_node(detail):
         max_end_date = max(x["end_date"] for x in detail if x["end_date"])
     except ValueError:
         max_end_date = None
+
+    dag_version_numbers = [
+        x.get("dag_version_number") for x in detail if x.get("dag_version_number") is not None
+    ]
+    dag_version_number = max(dag_version_numbers) if dag_version_numbers else None
+
     return {
         "state": agg_state(states),
         "min_start_date": min_start_date,
         "max_end_date": max_end_date,
         "child_states": dict(Counter(states)),
+        "dag_version_number": dag_version_number,
     }
 
 

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

@@ -8193,6 +8193,14 @@ export const $GridRunsResponse = {
         run_type: {
             '$ref': '#/components/schemas/DagRunType'
         },
+        dag_versions: {
+            items: {
+                '$ref': '#/components/schemas/DagVersionResponse'
+            },
+            type: 'array',
+            title: 'Dag Versions',
+            default: []
+        },
         has_missed_deadline: {
             type: 'boolean',
             title: 'Has Missed Deadline'
@@ -8308,6 +8316,17 @@ export const $LightGridTaskInstanceSummary = {
                 }
             ],
             title: 'Max End Date'
+        },
+        dag_version_number: {
+            anyOf: [
+                {
+                    type: 'integer'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Dag Version Number'
         }
     },
     type: 'object',

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

@@ -2009,6 +2009,7 @@ export type GridRunsResponse = {
     run_after: string;
     state: DagRunState | null;
     run_type: DagRunType;
+    dag_versions?: Array<DagVersionResponse>;
     has_missed_deadline: boolean;
     readonly duration: number;
 };
@@ -2043,6 +2044,7 @@ export type LightGridTaskInstanceSummary = {
 } | null;
     min_start_date: string | null;
     max_end_date: string | null;
+    dag_version_number?: number | null;
 };
 
 /**

airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json

@@ -121,6 +121,15 @@
     "graphDirection": {
       "label": "Graph Direction"
     },
+    "showVersionIndicator": {
+      "label": "Show Version Indicator",
+      "options": {
+        "hideAll": "Hide All",
+        "showAll": "Show All",
+        "showBundleVersion": "Show Bundle Version",
+        "showDagVersion": "Show Dag Version"
+      }
+    },
     "taskStreamFilter": {
       "activeFilter": "Active filter",
       "clearFilter": "Clear Filter",

airflow-core/src/airflow/ui/src/constants/localStorage.ts

@@ -26,6 +26,7 @@ export const CALENDAR_VIEW_MODE_KEY = "calendar-view-mode";
 export const LOG_WRAP_KEY = "log_wrap";
 export const LOG_SHOW_TIMESTAMP_KEY = "log_show_timestamp";
 export const LOG_SHOW_SOURCE_KEY = "log_show_source";
+export const VERSION_INDICATOR_DISPLAY_MODE_KEY = "version_indicator_display_mode";
 
 // Dag-scoped keys
 export const dagViewKey = (dagId: string) => `dag_view-${dagId}`;