Feat: Add version change indicators for Dag and bundle versions in Grid view

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

@@ -2183,6 +2183,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
@@ -2453,6 +2459,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

@@ -8143,6 +8143,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'
@@ -8258,6 +8266,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

@@ -1999,6 +1999,7 @@ export type GridRunsResponse = {
     run_after: string;
     state: DagRunState | null;
     run_type: DagRunType;
+    dag_versions?: Array<DagVersionResponse>;
     has_missed_deadline: boolean;
     readonly duration: number;
 };
@@ -2033,6 +2034,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}`;

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

@@ -0,0 +1,46 @@
+/*!
+ * 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.
+ */
+import { createListCollection } from "@chakra-ui/react";
+
+export enum VersionIndicatorOptions {
+  ALL = "all",
+  BUNDLE_VERSION = "bundle",
+  DAG_VERSION = "dag",
+  NONE = "none",
+}
+
+const validOptions = new Set<string>(Object.values(VersionIndicatorOptions));
+
+export const isVersionIndicatorOption = (value: unknown): value is VersionIndicatorOptions =>
+  typeof value === "string" && validOptions.has(value);
+
+export const showVersionIndicatorOptions = createListCollection({
+  items: [
+    { label: "dag:panel.showVersionIndicator.options.showAll", value: VersionIndicatorOptions.ALL },
+    {
+      label: "dag:panel.showVersionIndicator.options.showBundleVersion",
+      value: VersionIndicatorOptions.BUNDLE_VERSION,
+    },
+    {
+      label: "dag:panel.showVersionIndicator.options.showDagVersion",
+      value: VersionIndicatorOptions.DAG_VERSION,
+    },
+    { label: "dag:panel.showVersionIndicator.options.hideAll", value: VersionIndicatorOptions.NONE },
+  ],
+});

airflow-core/src/airflow/ui/src/layouts/Details/DetailsLayout.tsx

@@ -1,3 +1,5 @@
+/* eslint-disable max-lines */
+
 /*!
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -50,6 +52,7 @@ import {
   showGanttKey,
   triggeringUserFilterKey,
 } from "src/constants/localStorage";
+import { VersionIndicatorOptions } from "src/constants/showVersionIndicatorOptions";
 import { HoverProvider } from "src/context/hover";
 import { OpenGroupsProvider } from "src/context/openGroups";
 
@@ -88,6 +91,11 @@ export const DetailsLayout = ({ children, error, isLoading, tabs }: Props) => {
   );
 
   const [showGantt, setShowGantt] = useLocalStorage<boolean>(showGanttKey(dagId), false);
+  // Global setting: applies to all Dags (intentionally not scoped to dagId)
+  const [showVersionIndicatorMode, setShowVersionIndicatorMode] = useLocalStorage<VersionIndicatorOptions>(
+    `version_indicator_display_mode`,
+    VersionIndicatorOptions.ALL,
+  );
   const { fitView, getZoom } = useReactFlow();
   const { data: warningData } = useDagWarningServiceListDagWarnings({ dagId });
   const { onClose, onOpen, open } = useDisclosure();
@@ -161,8 +169,10 @@ export const DetailsLayout = ({ children, error, isLoading, tabs }: Props) => {
                   setLimit={setLimit}
                   setRunTypeFilter={setRunTypeFilter}
                   setShowGantt={setShowGantt}
+                  setShowVersionIndicatorMode={setShowVersionIndicatorMode}
                   setTriggeringUserFilter={setTriggeringUserFilter}
                   showGantt={showGantt}
+                  showVersionIndicatorMode={showVersionIndicatorMode}
                   triggeringUserFilter={triggeringUserFilter}
                 />
                 {dagView === "graph" ? (
@@ -174,6 +184,7 @@ export const DetailsLayout = ({ children, error, isLoading, tabs }: Props) => {
                       limit={limit}
                       runType={runTypeFilter}
                       showGantt={Boolean(runId) && showGantt}
+                      showVersionIndicatorMode={showVersionIndicatorMode}
                       triggeringUser={triggeringUserFilter}
                     />
                     {showGantt ? (

airflow-core/src/airflow/ui/src/layouts/Details/Grid/Bar.tsx

@@ -19,21 +19,27 @@
 import { Flex, Box } from "@chakra-ui/react";
 import { useParams, useSearchParams } from "react-router-dom";
 
-import type { GridRunsResponse } from "openapi/requests";
 import { RunTypeIcon } from "src/components/RunTypeIcon";
+import { VersionIndicatorOptions } from "src/constants/showVersionIndicatorOptions";
 import { useHover } from "src/context/hover";
 
 import { GridButton } from "./GridButton";
-
-const BAR_HEIGHT = 100;
+import { BundleVersionIndicator, DagVersionIndicator } from "./VersionIndicator";
+import { BAR_HEIGHT } from "./constants";
+import {
+  getBundleVersion,
+  getMaxVersionNumber,
+  type GridRunWithVersionFlags,
+} from "./useGridRunsWithVersionFlags";
 
 type Props = {
   readonly max: number;
   readonly onClick?: () => void;
-  readonly run: GridRunsResponse;
+  readonly run: GridRunWithVersionFlags;
+  readonly showVersionIndicatorMode?: VersionIndicatorOptions;
 };
 
-export const Bar = ({ max, onClick, run }: Props) => {
+export const Bar = ({ max, onClick, run, showVersionIndicatorMode }: Props) => {
   const { dagId = "", runId } = useParams();
   const [searchParams] = useSearchParams();
   const { hoveredRunId, setHoveredRunId } = useHover();
@@ -53,6 +59,17 @@ export const Bar = ({ max, onClick, run }: Props) => {
       position="relative"
       transition="background-color 0.2s"
     >
+      {run.isBundleVersionChange &&
+      (showVersionIndicatorMode === VersionIndicatorOptions.BUNDLE_VERSION ||
+        showVersionIndicatorMode === VersionIndicatorOptions.ALL) ? (
+        <BundleVersionIndicator bundleVersion={getBundleVersion(run)} />
+      ) : undefined}
+      {run.isDagVersionChange &&
+      (showVersionIndicatorMode === VersionIndicatorOptions.DAG_VERSION ||
+        showVersionIndicatorMode === VersionIndicatorOptions.ALL) ? (
+        <DagVersionIndicator dagVersionNumber={getMaxVersionNumber(run)} orientation="vertical" />
+      ) : undefined}
+
       <Flex
         alignItems="flex-end"
         height={BAR_HEIGHT}