UI: Add cross-Dag Time Schedule view

airflow-core/docs/ui.rst

@@ -571,3 +571,102 @@ Key pages include:
 ------------
 
 .. image:: img/ui-dark/admin_connections_add.png
+
+.. _ui-time-schedule:
+
+Time Schedule Views
+-------------------
+
+From the **Home** page, select the **Time Schedule** card beside the Stats section to view typical Dag start times and durations
+across the environment. Dag runs are grouped by time of day, making schedule patterns and congested periods easier to
+identify.
+
+.. image:: img/ui-dark/home_stats.png
+  :alt: Airflow Home Page showing link to Time Schedule (dark mode)
+
+|
+
+.. image:: img/ui-light/home_stats.png
+  :alt: Airflow Home Page showing link to Time Schedule (light mode)
+
+Day View
+''''''''
+
+The Day view places Dag IDs in rows and time of day on the horizontal axis. Overlapping bars use additional lanes.
+
+Use the **Dag ID** heading to sort by ascending ID, descending ID, or start time. Select a Dag ID to open the Dag.
+
+.. image:: img/ui-dark/time_schedule_day.png
+  :alt: Time Schedule Day view showing Dag runs by time of day (dark mode)
+
+|
+
+.. image:: img/ui-light/time_schedule_day.png
+  :alt: Time Schedule Day view showing Dag runs by time of day (light mode)
+
+Week View
+'''''''''
+
+The Week view places weekdays in columns and time of day on the vertical axis. It shows weekday patterns rather than a
+specific calendar week. Overlapping bars share the width of their weekday column.
+
+.. image:: img/ui-dark/time_schedule_week.png
+  :alt: Time Schedule Week view showing Dag runs by weekday and time of day (dark mode)
+
+|
+
+.. image:: img/ui-light/time_schedule_week.png
+  :alt: Time Schedule Week view showing Dag runs by weekday and time of day (light mode)
+
+Timeline Bars
+'''''''''''''
+
+Each bar represents one or more matching Dag runs at a similar time.
+
+Times use the selected UI timezone and each run's start time, falling back to **Run After** for runs that have not started.
+
+Bar colors and state icons indicate state:
+
+- Successful runs are green.
+- Failed runs are red.
+- Running runs are cyan.
+
+Available interactions:
+
+- Hover over a bar to see its Dag ID, state, timing, and aggregated Dag run count. The tooltip separates the Dag ID
+  from the run details and remains readable in light and dark themes.
+- Select a bar representing one Dag run to open that run.
+- Select an aggregated bar to open the Dag's **Runs** tab with **State**, **Run After**, and **Start Time** filters.
+  The **Week** view also applies **Start Weekday**. Matching runs outside the selected limit can also appear.
+
+Zoom controls
+'''''''''''''
+
+Use the **+** and **−** buttons to change the time scale. You can also use the mouse wheel or the Up and Down arrow
+keys while holding Ctrl on Windows and Linux, or Command on macOS.
+
+The Day view zooms the horizontal time axis. The Week view zooms the vertical time axis.
+
+The browser remembers the view, aggregation, and Dag run limit. Filters, including **Scheduled Dags only**, are stored
+in the page URL. The zoom level is not saved.
+
+Dag run limit
+'''''''''''''
+
+- **Limit 200** streams the 200 most recent matching Dag runs by default.
+- **Limit 600**, **Limit 1000**, **Limit 2000**, and **Limit 5000** increase the bounded number of runs
+  selected before aggregation.
+- The count above the timeline shows how many matching runs were rendered.
+- **Scheduled Dags only** is enabled by default and shows Dags with periodic timetables. Clear it to include other
+  Dags that have matching Dag runs.
+
+Duration aggregation
+''''''''''''''''''''
+
+Runs are grouped by Dag ID, state, and time of day. The Week view also groups runs by weekday.
+
+The timeline marker interval sets the aggregation window. Zoom in to use a more precise window.
+
+- **Average duration**: average start and end times (default)
+- **Full time range**: earliest start and latest end times
+- **Shortest run**: start and end times of the shortest Dag run

airflow-core/newsfragments/71558.feature.rst

@@ -0,0 +1 @@
+Add Time Schedule views to compare Dag run timing, duration, and state across the environment.

airflow-core/src/airflow/api_fastapi/common/parameters/__init__.py

@@ -78,6 +78,7 @@
     QueryDagRunPartitionKeyPrefixSearch as QueryDagRunPartitionKeyPrefixSearch,
     QueryDagRunPartitionKeySearch as QueryDagRunPartitionKeySearch,
     QueryDagRunRunTypesFilter as QueryDagRunRunTypesFilter,
+    QueryDagRunStartWeekdayFilter as QueryDagRunStartWeekdayFilter,
     QueryDagRunStateFilter as QueryDagRunStateFilter,
     QueryDagRunTriggeringUserPrefixSearch as QueryDagRunTriggeringUserPrefixSearch,
     QueryDagRunTriggeringUserSearch as QueryDagRunTriggeringUserSearch,
@@ -129,6 +130,7 @@
     datetime_range_filter_factory as datetime_range_filter_factory,
     float_range_filter_factory as float_range_filter_factory,
     int_range_filter_factory as int_range_filter_factory,
+    time_range_filter_factory as time_range_filter_factory,
 )
 from airflow.api_fastapi.common.parameters.search import (
     _PrefixSearchParam as _PrefixSearchParam,

airflow-core/src/airflow/api_fastapi/common/parameters/dag_run.py

@@ -18,17 +18,18 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
-from datetime import timedelta
+from datetime import datetime, timedelta
 from typing import (
     TYPE_CHECKING,
     Annotated,
 )
 
 from fastapi import Depends, HTTPException, Query
-from sqlalchemy import select as sql_select
+from sqlalchemy import extract, func, select as sql_select
 from sqlalchemy.orm import aliased
 
 from airflow._shared.timezones import timezone
+from airflow.api_fastapi.common.db.common import SessionDep
 from airflow.api_fastapi.common.parameters.base import BaseParam
 from airflow.api_fastapi.common.parameters.filter import FilterOptionEnum, FilterParam, filter_param_factory
 from airflow.api_fastapi.common.parameters.search import (
@@ -45,14 +46,49 @@
 from airflow.utils.types import DagRunType
 
 if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
     from sqlalchemy.sql import Select
+    from sqlalchemy.sql.elements import ColumnElement
 
 
 # A lookback this large is effectively unbounded (users omit the param for "any time"); capping it
 # also keeps utcnow() - timedelta(hours=...) from overflowing on an absurdly large value.
 _MAX_DAG_RUN_STATE_WINDOW_HOURS = 24 * 366 * 100  # ~100 years
 
 
+class _WeekdayFilter(BaseParam[list[int]]):
+    """Filter DagRuns by selected UTC weekdays."""
+
+    def __init__(self, session: Session, timestamp: ColumnElement[datetime], value: list[int] | None) -> None:
+        super().__init__(value)
+        self.session = session
+        self.timestamp = timestamp
+
+    def to_orm(self, select: Select) -> Select:
+        if not self.value:
+            return select
+        weekday = (
+            func.dayofweek(self.timestamp) - 1
+            if self.session.get_bind().dialect.name == "mysql"
+            else extract("dow", self.timestamp)
+        )
+        return select.where(weekday.in_(self.value))
+
+    @classmethod
+    def depends(
+        cls,
+        session: SessionDep,
+        start_weekday: list[Annotated[int, Query(ge=0, le=6)]] | None = Query(
+            None, description="Match any selected weekday. Sunday=0, Saturday=6."
+        ),
+    ) -> _WeekdayFilter:
+        timestamp = func.coalesce(DagRun.start_date, DagRun.run_after)
+        return cls(session, timestamp, start_weekday)
+
+
+QueryDagRunStartWeekdayFilter = Annotated[_WeekdayFilter, Depends(_WeekdayFilter.depends)]
+
+
 class _AnyDagRunStateFilter(BaseParam[DagRunState | None]):
     """Filter Dags that have any DagRun in the given state, not only the latest one."""
 

airflow-core/src/airflow/api_fastapi/common/parameters/range.py

@@ -18,7 +18,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from datetime import datetime
+from datetime import datetime, time
 from typing import (
     TYPE_CHECKING,
     Annotated,
@@ -27,10 +27,10 @@
     overload,
 )
 
-from fastapi import HTTPException, Query
+from fastapi import HTTPException, Query, status
 from pendulum.parsing.exceptions import ParserError
 from pydantic import AfterValidator, BaseModel
-from sqlalchemy import and_, func, or_
+from sqlalchemy import and_, extract, func, or_
 
 from airflow._shared.timezones import timezone
 from airflow.api_fastapi.common.parameters.base import BaseParam, T
@@ -39,7 +39,7 @@
 
 if TYPE_CHECKING:
     from sqlalchemy.orm.attributes import InstrumentedAttribute
-    from sqlalchemy.sql import Select
+    from sqlalchemy.sql import ColumnElement, Select
 
 
 def _safe_parse_datetime(date_to_check: str) -> datetime:
@@ -91,9 +91,11 @@ class Range(BaseModel, Generic[T]):
 class RangeFilter(BaseParam[Range]):
     """Filter on range in between the lower and upper bound."""
 
-    def __init__(self, value: Range | None, attribute: InstrumentedAttribute) -> None:
+    def __init__(
+        self, value: Range | None, attribute: ColumnElement[Any] | InstrumentedAttribute[Any]
+    ) -> None:
         super().__init__(value)
-        self.attribute: InstrumentedAttribute = attribute
+        self.attribute: ColumnElement[Any] | InstrumentedAttribute[Any] = attribute
 
     def to_orm(self, select: Select) -> Select:
         if self.skip_none is False:
@@ -102,13 +104,13 @@ def to_orm(self, select: Select) -> Select:
         if self.value is None:
             return select
 
-        if self.value.lower_bound_gte:
+        if self.value.lower_bound_gte is not None:
             select = select.where(self.attribute >= self.value.lower_bound_gte)
-        if self.value.lower_bound_gt:
+        if self.value.lower_bound_gt is not None:
             select = select.where(self.attribute > self.value.lower_bound_gt)
-        if self.value.upper_bound_lte:
+        if self.value.upper_bound_lte is not None:
             select = select.where(self.attribute <= self.value.upper_bound_lte)
-        if self.value.upper_bound_lt:
+        if self.value.upper_bound_lt is not None:
             select = select.where(self.attribute < self.value.upper_bound_lt)
 
         return select
@@ -188,6 +190,48 @@ def depends_datetime(
     return depends_datetime
 
 
+def time_range_filter_factory(filter_name: str, model: Base) -> Callable[..., RangeFilter]:
+    """Create a dependency for filtering DagRuns by the UTC time of their start date."""
+    timestamp = func.coalesce(getattr(model, "start_date"), getattr(model, "run_after"))
+
+    def depends_time_range(
+        start_time_gte: time | None = Query(None, alias=f"{filter_name}_gte"),
+        start_time_lt: time | None = Query(None, alias=f"{filter_name}_lt"),
+    ) -> RangeFilter:
+        def to_utc_minutes(value: time | None, default: int) -> int:
+            if value is None:
+                return default
+            offset = value.utcoffset()
+            if offset is None:
+                raise HTTPException(
+                    status.HTTP_422_UNPROCESSABLE_CONTENT, "Time bounds must include a timezone."
+                )
+            offset_minutes = int(offset.total_seconds() // 60)
+            return (value.hour * 60 + value.minute - offset_minutes) % (24 * 60)
+
+        lower = to_utc_minutes(start_time_gte, 0)
+        upper = to_utc_minutes(start_time_lt, 24 * 60)
+        if start_time_gte is not None and start_time_lt is not None and lower == upper:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST, "start_time_gte must be earlier than start_time_lt."
+            )
+        minute = extract("hour", timestamp) * 60 + extract("minute", timestamp)
+        if start_time_gte is not None and start_time_lt is not None and lower > upper:
+            # A local daytime range can cross midnight when converted to UTC.
+            minute = (minute - lower + 24 * 60) % (24 * 60)
+            upper = (upper - lower) % (24 * 60)
+            lower = 0
+        value = Range(
+            lower_bound_gte=lower if start_time_gte is not None else None,
+            lower_bound_gt=None,
+            upper_bound_lte=None,
+            upper_bound_lt=upper if start_time_lt is not None else None,
+        )
+        return RangeFilter(value, minute)
+
+    return depends_time_range
+
+
 def float_range_filter_factory(
     filter_name: str, model: Base
 ) -> Callable[[float | None, float | None, float | None, float | None], RangeFilter]:

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

@@ -0,0 +1,50 @@
+# 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.
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.utils.state import DagRunState
+
+
+class TimeScheduleItem(BaseModel):
+    """An aggregated bar in the Time Schedule UI."""
+
+    dag_id: str
+    dag_run_id: str
+    duration_ms: float
+    end_date: datetime | None
+    is_time_scheduled: bool
+    dag_display_name: str
+    run_count: int
+    # Preserve the real date range when bars combine runs from different days.
+    run_after_min: datetime
+    run_after_max: datetime
+    start_date: datetime | None
+    state: DagRunState
+    start_time_gte: str | None = None
+    start_time_lt: str | None = None
+    start_weekday: int | None = None
+
+
+class TimeScheduleBatch(BaseModel):
+    """A progressively streamed batch of Time Schedule bars."""
+
+    dag_run_count: int
+    items: list[TimeScheduleItem]