apache · jedcunningham · Jul 18, 2026
@@ -52,7 +52,9 @@ class HistoricalMetricDataResponse(BaseModel):
 
     dag_run_states: DAGRunStates
     task_instance_states: TaskInstanceStateCount
-    state_count_limit: int
+    # True when the counts above are floors on the real values rather than exact figures.
+    dag_run_counts_are_lower_bounds: bool = False
+    task_instance_counts_are_lower_bounds: bool = False
 
 
 class DashboardDagStatsResponse(BaseModel):

@@ -3763,14 +3763,18 @@ components:
           $ref: '#/components/schemas/DAGRunStates'
         task_instance_states:
           $ref: '#/components/schemas/TaskInstanceStateCount'
-        state_count_limit:
-          type: integer
-          title: State Count Limit
+        dag_run_counts_are_lower_bounds:
+          type: boolean
+          title: Dag Run Counts Are Lower Bounds
+          default: false
+        task_instance_counts_are_lower_bounds:
+          type: boolean
+          title: Task Instance Counts Are Lower Bounds
+          default: false
       type: object
       required:
       - dag_run_states
       - task_instance_states
-      - state_count_limit
       title: HistoricalMetricDataResponse
       description: Historical Metric Data serializer for responses.
     JobResponse:

@@ -68,7 +68,6 @@
     DAGWithLatestDagRunsResponse,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
-from airflow.api_fastapi.core_api.routes.ui.dashboard import STATE_COUNT_CAP
 from airflow.api_fastapi.core_api.security import (
     GetUserDep,
     ReadableDagsFilterDep,
@@ -83,6 +82,9 @@
 
 dags_router = AirflowRouter(prefix="/dags", tags=["DAG"])
 
+# Per-dag run counts read at most this many rows per state; the UI shows "N+" at the cap.
+STATE_COUNT_CAP = 1000
+
 
 @dags_router.get(
     "",

@@ -16,10 +16,11 @@
 # under the License.
 from __future__ import annotations
 
-from typing import cast
+from decimal import ROUND_FLOOR, Context
+from typing import TYPE_CHECKING, cast
 
 from fastapi import Depends, status
-from sqlalchemy import func, literal, select, union_all
+from sqlalchemy import func, select
 from sqlalchemy.sql.expression import case, false
 
 from airflow._shared.timezones import timezone
@@ -38,11 +39,45 @@
 from airflow.models.taskinstance import TaskInstance
 from airflow.utils.state import DagRunState, TaskInstanceState
 
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
 dashboard_router = AirflowRouter(tags=["Dashboard"], prefix="/dashboard")
 
-# Cap for state counts — avoids counting millions of rows.
-# The UI shows "N+" when the returned count equals this value.
-STATE_COUNT_CAP = 1000
+# Rows a single scan reads. Windows that fit are counted exactly; wider ones report a floor.
+EXACT_COUNT_LIMIT = 50_000
+
+
+_ROUNDING = Context(prec=2, rounding=ROUND_FLOOR)
+
+
+def _round_down(value: int) -> int:
+    """Round to two significant digits, never upwards: that would claim uncounted rows."""
+    return int(_ROUNDING.create_decimal(value))
+
+
+def _compute_state_counts(
+    model, filters, *, session: Session, join=None, null_label: str | None = None
+) -> tuple[dict[str, int], bool]:
+    """
+    Per-state counts for the window, and whether they are lower bounds rather than exact.
+
+    A scan that stopped early counted only some of the rows, but each count is still a floor
+    on the real value.
+    """
+    stmt = select(model.state.label("state")).select_from(model)
+    if join is not None:
+        stmt = stmt.join(join)
+    window = stmt.where(*filters).limit(EXACT_COUNT_LIMIT + 1).subquery()
+    rows = session.execute(select(window.c.state, func.count().label("cnt")).group_by(window.c.state)).all()
+    are_lower_bounds = sum(row.cnt for row in rows) > EXACT_COUNT_LIMIT
+    counts: dict[str, int] = {}
+    for row in rows:
+        label = row.state or null_label
+        if label is None:
+            continue
+        counts[label] = _round_down(row.cnt) if are_lower_bounds else row.cnt
+    return counts, are_lower_bounds
 
 
 @dashboard_router.get(
@@ -70,50 +105,31 @@ def historical_metrics(
         DagRun.dag_id.in_(permitted_dag_ids),
     ]
 
-    # Build one LIMIT-capped subquery per state, then UNION ALL them into a
-    # single query.  Every state gets the same treatment: at most STATE_COUNT_CAP
-    # rows are read from the index, so even states with millions of rows
-    # (typically "success") are counted in single-digit milliseconds.
-    # Each branch is wrapped in a subquery so LIMIT works on all backends
-    # (SQLite rejects LIMIT inside bare UNION ALL arms).
-    def _capped_state_counts(model, states, label_fn, join=None):
-        branches = []
-        for state in states:
-            stmt = select(literal(label_fn(state)).label("state")).select_from(model)
-            if join is not None:
-                stmt = stmt.join(join)
-            branch = (
-                stmt.where(*dag_run_filters)
-                .where(model.state == state if state else model.state.is_(None))
-                .limit(STATE_COUNT_CAP)
-                .subquery()
-            )
-            branches.append(select(branch.c.state))
-        capped = union_all(*branches).subquery()
-        return session.execute(
-            select(capped.c.state, func.count().label("cnt")).group_by(capped.c.state)
-        ).all()
-
-    dag_run_state_counts = _capped_state_counts(DagRun, list(DagRunState), lambda s: s.value)
-    ti_state_counts = _capped_state_counts(
+    # Judged separately: dag runs often fit when task instances do not.
+    dag_run_states, dag_runs_are_lower_bounds = _compute_state_counts(
+        DagRun, dag_run_filters, session=session
+    )
+    task_instance_states, task_instances_are_lower_bounds = _compute_state_counts(
         TaskInstance,
-        [None, *TaskInstanceState],
-        lambda s: s.value if s else "no_status",
+        dag_run_filters,
+        session=session,
         join=TaskInstance.dag_run,
+        null_label="no_status",
     )
 
     return HistoricalMetricDataResponse.model_validate(
         {
             "dag_run_states": {
                 **{dag_run_state.value: 0 for dag_run_state in DagRunState},
-                **{row.state: row.cnt for row in dag_run_state_counts},
+                **dag_run_states,
             },
             "task_instance_states": {
                 "no_status": 0,
                 **{ti_state.value: 0 for ti_state in TaskInstanceState},
-                **{row.state: row.cnt for row in ti_state_counts},
+                **task_instance_states,
             },
-            "state_count_limit": STATE_COUNT_CAP,
+            "dag_run_counts_are_lower_bounds": dag_runs_are_lower_bounds,
+            "task_instance_counts_are_lower_bounds": task_instances_are_lower_bounds,
         }
     )
 

@@ -10165,13 +10165,19 @@ export const $HistoricalMetricDataResponse = {
         task_instance_states: {
             '$ref': '#/components/schemas/TaskInstanceStateCount'
         },
-        state_count_limit: {
-            type: 'integer',
-            title: 'State Count Limit'
+        dag_run_counts_are_lower_bounds: {
+            type: 'boolean',
+            title: 'Dag Run Counts Are Lower Bounds',
+            default: false
+        },
+        task_instance_counts_are_lower_bounds: {
+            type: 'boolean',
+            title: 'Task Instance Counts Are Lower Bounds',
+            default: false
         }
     },
     type: 'object',
-    required: ['dag_run_states', 'task_instance_states', 'state_count_limit'],
+    required: ['dag_run_states', 'task_instance_states'],
     title: 'HistoricalMetricDataResponse',
     description: 'Historical Metric Data serializer for responses.'
 } as const;

@@ -2586,7 +2586,8 @@ export type GridTISummaries = {
 export type HistoricalMetricDataResponse = {
     dag_run_states: DAGRunStates;
     task_instance_states: TaskInstanceStateCount;
-    state_count_limit: number;
+    dag_run_counts_are_lower_bounds?: boolean;
+    task_instance_counts_are_lower_bounds?: boolean;
 };
 
 /**

@@ -24,21 +24,26 @@ import { FiBarChart } from "react-icons/fi";
 import { MetricSection } from "./MetricSection";
 
 type DagRunMetricsProps = {
+  readonly countsAreLowerBounds: boolean;
   readonly dagRunStates: DAGRunStates;
   readonly endDate?: string;
   readonly startDate: string;
-  readonly stateCountLimit: number;
 };
 
 const DAGRUN_STATES: Array<keyof DAGRunStates> = ["queued", "running", "success", "failed"];
 
-export const DagRunMetrics = ({ dagRunStates, endDate, startDate, stateCountLimit }: DagRunMetricsProps) => {
+export const DagRunMetrics = ({
+  countsAreLowerBounds,
+  dagRunStates,
+  endDate,
+  startDate,
+}: DagRunMetricsProps) => {
   const { t: translate } = useTranslation();
   const total = Object.values(dagRunStates).reduce((sum, count) => sum + count, 0);
-  // When any state hit the API's STATE_COUNT_CAP, the summed total is only a
-  // lower bound, so per-state percentages computed from it are wrong (#67336).
-  // Suppress percentages for the whole group in that case.
-  const isTotalTruncated = Object.values(dagRunStates).some((count) => count >= stateCountLimit);
+  // The total is only a lower bound when the counts are, so percentages would be wrong.
+  const isTotalTruncated = countsAreLowerBounds;
+  // "0+" would be meaningless.
+  const isLowerBound = (count: number) => countsAreLowerBounds && count > 0;
 
   return (
     <Box borderRadius={5} borderWidth={1} p={4}>
@@ -50,7 +55,7 @@ export const DagRunMetrics = ({ dagRunStates, endDate, startDate, stateCountLimi
       <Stack gap={4}>
         {DAGRUN_STATES.map((state) => (
           <MetricSection
-            capped={dagRunStates[state] >= stateCountLimit}
+            capped={isLowerBound(dagRunStates[state])}
             endDate={endDate}
             isTotalTruncated={isTotalTruncated}
             key={state}

@@ -74,13 +74,13 @@ export const HistoricalMetrics = ({ endDate, startDate }: HistoricalMetricsProps
             {!isLoading && data !== undefined && (
               <Box>
                 <DagRunMetrics
+                  countsAreLowerBounds={data.dag_run_counts_are_lower_bounds ?? false}
                   dagRunStates={data.dag_run_states}
                   startDate={startDate}
-                  stateCountLimit={data.state_count_limit}
                 />
                 <TaskInstanceMetrics
+                  countsAreLowerBounds={data.task_instance_counts_are_lower_bounds ?? false}
                   startDate={startDate}
-                  stateCountLimit={data.state_count_limit}
                   taskInstanceStates={data.task_instance_states}
                 />
               </Box>

@@ -48,6 +48,7 @@ export const MetricSection = ({
   state,
   total,
 }: MetricSectionProps) => {
+  // A lower bound has no known proportion, so it deliberately fills the bar.
   const stateWidth = capped ? BAR_WIDTH : total === 0 ? 0 : (runs / total) * BAR_WIDTH;
   const remainingWidth = BAR_WIDTH - stateWidth;
   const hidePercent = isTotalTruncated;
@@ -57,7 +58,7 @@ export const MetricSection = ({
   const searchParams = new URLSearchParams(
     `?${stateParam}=${state}&${SearchParamsKeys.START_DATE_GTE}=${startDate}`,
   );
-  const { t: translate } = useTranslation();
+  const { i18n, t: translate } = useTranslation();
 
   if (endDate !== undefined) {
     searchParams.append(SearchParamsKeys.END_DATE, endDate);
@@ -70,7 +71,7 @@ export const MetricSection = ({
           <RouterLink to={`/${kind}?${searchParams.toString()}`}>
             <StateBadge fontSize="md" state={state === "no_status" ? null : state}>
               {}
-              {capped ? `${runs}+` : runs}
+              {`${runs.toLocaleString(i18n.language)}${capped ? "+" : ""}`}
             </StateBadge>
           </RouterLink>
           <Text>{translate(`states.${state}`)}</Text>

@@ -24,9 +24,9 @@ import { MdOutlineTask } from "react-icons/md";
 import { MetricSection } from "./MetricSection";
 
 type TaskInstanceMetricsProps = {
+  readonly countsAreLowerBounds: boolean;
   readonly endDate?: string;
   readonly startDate: string;
-  readonly stateCountLimit: number;
   readonly taskInstanceStates: TaskInstanceStateCount;
 };
 
@@ -48,17 +48,17 @@ const TASK_STATES: Array<keyof TaskInstanceStateCount> = [
 ];
 
 export const TaskInstanceMetrics = ({
+  countsAreLowerBounds,
   endDate,
   startDate,
-  stateCountLimit,
   taskInstanceStates,
 }: TaskInstanceMetricsProps) => {
   const { t: translate } = useTranslation();
   const total = Object.values(taskInstanceStates).reduce((sum, count) => sum + count, 0);
-  // When any state hit the API's STATE_COUNT_CAP, the summed total is only a
-  // lower bound, so per-state percentages computed from it are wrong (#67336).
-  // Suppress percentages for the whole group in that case.
-  const isTotalTruncated = Object.values(taskInstanceStates).some((count) => count >= stateCountLimit);
+  // The total is only a lower bound when the counts are, so percentages would be wrong.
+  const isTotalTruncated = countsAreLowerBounds;
+  // "0+" would be meaningless.
+  const isLowerBound = (count: number) => countsAreLowerBounds && count > 0;
 
   return (
     <Box borderRadius={5} borderWidth={1} mt={2} p={4}>
@@ -73,7 +73,7 @@ export const TaskInstanceMetrics = ({
         ).map((state) =>
           taskInstanceStates[state] > 0 ? (
             <MetricSection
-              capped={taskInstanceStates[state] >= stateCountLimit}
+              capped={isLowerBound(taskInstanceStates[state])}
               endDate={endDate}
               isTotalTruncated={isTotalTruncated}
               key={state}