This is an automated email from the ASF dual-hosted git repository.
bbovenzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e4d0382fa18 Restore count to Dag Run and Task Instance lists (#71252)
e4d0382fa18 is described below
commit e4d0382fa1877c6b5185c6ed2b79c02c04fac23d
Author: Brent Bovenzi <[email protected]>
AuthorDate: Wed Aug 12 16:05:40 2026 -0400
Restore count to Dag Run and Task Instance lists (#71252)
* add bounds to dag run and TI lists
* Fix CI
---
.../src/airflow/api_fastapi/common/db/common.py | 30 +++++++++++++++
.../api_fastapi/core_api/datamodels/dag_run.py | 10 ++++-
.../core_api/datamodels/task_instances.py | 10 ++++-
.../core_api/openapi/v2-rest-api-generated.yaml | 44 ++++++++++++++++------
.../api_fastapi/core_api/routes/public/dag_run.py | 15 +++++++-
.../core_api/routes/public/task_instances.py | 19 ++++++++--
.../api_fastapi/core_api/routes/ui/dashboard.py | 5 +--
.../ui/openapi-gen/queries/ensureQueryData.ts | 10 +++--
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 10 +++--
.../src/airflow/ui/openapi-gen/queries/queries.ts | 10 +++--
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 10 +++--
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 28 +++++++++++++-
.../ui/openapi-gen/requests/services.gen.ts | 10 +++--
.../airflow/ui/openapi-gen/requests/types.gen.ts | 12 +++++-
.../ui/src/components/DataTable/DataTable.test.tsx | 36 ++++++++++++++++++
.../ui/src/components/DataTable/DataTable.tsx | 12 ++++--
.../ui/src/pages/Dag/Backfills/Backfills.test.tsx | 1 +
.../src/airflow/ui/src/pages/DagRuns/DagRuns.tsx | 2 +
.../ui/src/pages/TaskInstances/TaskInstances.tsx | 2 +
.../core_api/routes/public/test_dag_run.py | 31 ++++++++++++---
.../core_api/routes/public/test_task_instances.py | 22 ++++++++---
.../src/airflowctl/api/datamodels/generated.py | 18 ++++++++-
22 files changed, 282 insertions(+), 65 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/db/common.py
b/airflow-core/src/airflow/api_fastapi/common/db/common.py
index ff1e3f7043e..8abcaf59fa8 100644
--- a/airflow-core/src/airflow/api_fastapi/common/db/common.py
+++ b/airflow-core/src/airflow/api_fastapi/common/db/common.py
@@ -26,6 +26,7 @@ from collections.abc import AsyncGenerator, Generator,
Sequence
from typing import TYPE_CHECKING, Annotated, Literal, overload
from fastapi import Depends
+from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session
@@ -37,6 +38,11 @@ if TYPE_CHECKING:
from airflow.api_fastapi.core_api.base import OrmClause
+# Rows a single scan reads. Result sets that fit are counted exactly; wider
ones report a floor.
+# Shared by the dashboard's historical metrics and by cursor-paginated
listings, which surface an
+# item count without counting every matching row — the point of cursor
pagination on large tables.
+EXACT_COUNT_LIMIT = 50_000
+
def _get_session() -> Generator[Session, None, None]:
with create_session(scoped=False) as session:
@@ -59,6 +65,30 @@ def apply_filters_to_select(
return statement
+def bounded_total_entries(
+ *,
+ statement: Select,
+ filters: Sequence[OrmClause | None] | None = None,
+ session: Session,
+) -> tuple[int, int]:
+ """
+ Count the rows a cursor-paginated listing matches, reading at most
``EXACT_COUNT_LIMIT``.
+
+ Returns ``(total, limit)`` where ``total`` is ``min(actual_count,
EXACT_COUNT_LIMIT)`` — a
+ ``total`` equal to ``limit`` means only that at least that many rows match
— and ``limit`` is
+ the cap that was applied, for the caller to surface as
``total_entries_limit``.
+
+ The ``LIMIT`` sits inside the counted subquery so the database stops
scanning once the cap is
+ reached, keeping the count cheap on tables that cursor pagination exists
to handle. ORDER BY is
+ stripped for the same reason :func:`~airflow.utils.db.get_query_count`
strips it: it cannot
+ change a count and only constrains the planner.
+ """
+ statement = apply_filters_to_select(statement=statement, filters=filters)
+ bounded = statement.order_by(None).limit(EXACT_COUNT_LIMIT).subquery()
+ total = session.scalar(select(func.count()).select_from(bounded)) or 0
+ return total, EXACT_COUNT_LIMIT
+
+
async def _get_async_session() -> AsyncGenerator[AsyncSession, None]:
async with create_session_async() as session:
yield session
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
index d728f8fd0b0..1b1dd37eb9e 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
@@ -202,8 +202,14 @@ class DAGRunCollectionResponse(BaseModel):
dag_runs: Iterable[DAGRunResponse]
total_entries: int | None = Field(
default=None,
- description="Total number of matching items. Populated for offset
pagination, "
- "``null`` when using cursor pagination.",
+ description="Number of matching items. For offset pagination this is
the exact total. "
+ "For cursor pagination it is capped at ``total_entries_limit``; a
value equal to that "
+ "limit means at least that many items match.",
+ )
+ total_entries_limit: int | None = Field(
+ default=None,
+ description="Cap applied to ``total_entries`` under cursor pagination.
``null`` for offset "
+ "pagination, where ``total_entries`` is exact.",
)
next_cursor: str | None = Field(
default=None,
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
index 4ea8a3c1f72..7166835ce61 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py
@@ -107,8 +107,14 @@ class TaskInstanceCollectionResponse(BaseModel):
task_instances: Iterable[TaskInstanceResponse]
total_entries: int | None = Field(
default=None,
- description="Total number of matching items. Populated for offset
pagination, "
- "``null`` when using cursor pagination.",
+ description="Number of matching items. For offset pagination this is
the exact total. "
+ "For cursor pagination it is capped at ``total_entries_limit``; a
value equal to that "
+ "limit means at least that many items match.",
+ )
+ total_entries_limit: int | None = Field(
+ default=None,
+ description="Cap applied to ``total_entries`` under cursor pagination.
``null`` for offset "
+ "pagination, where ``total_entries`` is exact.",
)
next_cursor: str | None = Field(
default=None,
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index b3ed8a3a2b8..5200635ec9e 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -2270,8 +2270,11 @@ paths:
**Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor`
from the response).
- When `cursor` is provided, `offset` is ignored and `total_entries` is
not
- returned.
+ When `cursor` is provided, `offset` is ignored and `total_entries` is
capped
+ at
+
+ `total_entries_limit` (a value equal to that limit means at least that
many
+ runs match).
``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor``
is ``null``
@@ -8175,13 +8178,16 @@ paths:
**Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor`
from the response).
- When `cursor` is provided, `offset` is ignored and `total_entries` is
not
- returned.
+ When `cursor` is provided, `offset` is ignored and `total_entries` is
capped
+ at
- ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor``
- is ``null``
+ `total_entries_limit` (a value equal to that limit means at least that
many
+ task instances
- on the first page.'
+ match). ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor``
+ is
+
+ ``null`` on the first page.'
operationId: get_task_instances
security:
- OAuth2PasswordBearer: []
@@ -13410,8 +13416,16 @@ components:
- type: integer
- type: 'null'
title: Total Entries
- description: Total number of matching items. Populated for offset
pagination,
- ``null`` when using cursor pagination.
+ description: Number of matching items. For offset pagination this is
the
+ exact total. For cursor pagination it is capped at
``total_entries_limit``;
+ a value equal to that limit means at least that many items match.
+ total_entries_limit:
+ anyOf:
+ - type: integer
+ - type: 'null'
+ title: Total Entries Limit
+ description: Cap applied to ``total_entries`` under cursor
pagination. ``null``
+ for offset pagination, where ``total_entries`` is exact.
next_cursor:
anyOf:
- type: string
@@ -15422,8 +15436,16 @@ components:
- type: integer
- type: 'null'
title: Total Entries
- description: Total number of matching items. Populated for offset
pagination,
- ``null`` when using cursor pagination.
+ description: Number of matching items. For offset pagination this is
the
+ exact total. For cursor pagination it is capped at
``total_entries_limit``;
+ a value equal to that limit means at least that many items match.
+ total_entries_limit:
+ anyOf:
+ - type: integer
+ - type: 'null'
+ title: Total Entries Limit
+ description: Cap applied to ``total_entries`` under cursor
pagination. ``null``
+ for offset pagination, where ``total_entries`` is exact.
next_cursor:
anyOf:
- type: string
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index f2310f092d5..d43e3b56fd8 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -37,7 +37,12 @@ from airflow.api_fastapi.common.cursors import (
parse_cursor,
)
from airflow.api_fastapi.common.dagbag import DagBagDep, get_dag_for_run,
get_latest_version_of_dag
-from airflow.api_fastapi.common.db.common import SessionDep,
apply_filters_to_select, paginated_select
+from airflow.api_fastapi.common.db.common import (
+ SessionDep,
+ apply_filters_to_select,
+ bounded_total_entries,
+ paginated_select,
+)
from airflow.api_fastapi.common.db.dag_runs import (
attach_dag_versions_to_runs,
eager_load_dag_run_for_list,
@@ -587,7 +592,8 @@ def get_dag_runs(
**Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
**Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
- When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+ When `cursor` is provided, `offset` is ignored and `total_entries` is
capped at
+ `total_entries_limit` (a value equal to that limit means at least that
many runs match).
``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
on the first page.
"""
@@ -696,8 +702,13 @@ def get_dag_runs(
attach_dag_versions_to_runs(dag_runs, session=session)
+ total_entries, total_entries_limit = bounded_total_entries(
+ statement=query, filters=filters, session=session
+ )
return DAGRunCollectionResponse(
dag_runs=dag_runs,
+ total_entries=total_entries,
+ total_entries_limit=total_entries_limit,
next_cursor=(encode_cursor(dag_runs[-1], order_by) if has_next and
dag_runs else None),
previous_cursor=(
make_backward_cursor(encode_cursor(dag_runs[0], order_by)) if
has_prev and dag_runs else None
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
index b5558746ff0..6ccb01dd68d 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
@@ -40,7 +40,12 @@ from airflow.api_fastapi.common.dagbag import (
get_latest_version_of_dag,
resolve_run_on_latest_version,
)
-from airflow.api_fastapi.common.db.common import SessionDep,
apply_filters_to_select, paginated_select
+from airflow.api_fastapi.common.db.common import (
+ SessionDep,
+ apply_filters_to_select,
+ bounded_total_entries,
+ paginated_select,
+)
from airflow.api_fastapi.common.db.dags import eager_load_teams
from airflow.api_fastapi.common.db.task_instances import
eager_load_TI_and_TIH_for_validation
from airflow.api_fastapi.common.parameters import (
@@ -547,9 +552,10 @@ def get_task_instances(
**Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
**Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
- When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
- ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
- on the first page.
+ When `cursor` is provided, `offset` is ignored and `total_entries` is
capped at
+ `total_entries_limit` (a value equal to that limit means at least that
many task instances
+ match). ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is
+ ``null`` on the first page.
"""
use_cursor = cursor is not None
dag_run = None
@@ -642,8 +648,13 @@ def get_task_instances(
has_prev = bool(cursor)
has_next = has_more
+ total_entries, total_entries_limit = bounded_total_entries(
+ statement=query, filters=filters, session=session
+ )
return TaskInstanceCollectionResponse(
task_instances=task_instances,
+ total_entries=total_entries,
+ total_entries_limit=total_entries_limit,
next_cursor=(
encode_cursor(task_instances[-1], order_by) if has_next and
task_instances else None
),
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
index 62c51c79d2e..483cbc009dc 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dashboard.py
@@ -25,7 +25,7 @@ from sqlalchemy.sql.expression import case, false
from airflow._shared.timezones import timezone
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity
-from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.db.common import EXACT_COUNT_LIMIT, SessionDep
from airflow.api_fastapi.common.parameters import DateTimeQuery,
OptionalDateTimeQuery
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.datamodels.ui.dashboard import (
@@ -44,9 +44,6 @@ if TYPE_CHECKING:
dashboard_router = AirflowRouter(tags=["Dashboard"], prefix="/dashboard")
-# 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)
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index c7d379c85d9..8f7ed508d3a 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -341,7 +341,8 @@ export const ensureUseDagRunServiceGetDagRunData =
(queryClient: QueryClient, {
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
runs match).
* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
* on the first page.
* @param data The data for the request.
@@ -1105,9 +1106,10 @@ export const
ensureUseTaskInstanceServiceGetMappedTaskInstanceData = (queryClien
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
-* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
-* on the first page.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
task instances
+* match). ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is
+* ``null`` on the first page.
* @param data The data for the request.
* @param data.dagId
* @param data.dagRunId
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index f0af02f94f6..fb50dcc1247 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -341,7 +341,8 @@ export const prefetchUseDagRunServiceGetDagRun =
(queryClient: QueryClient, { da
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
runs match).
* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
* on the first page.
* @param data The data for the request.
@@ -1105,9 +1106,10 @@ export const
prefetchUseTaskInstanceServiceGetMappedTaskInstance = (queryClient:
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
-* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
-* on the first page.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
task instances
+* match). ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is
+* ``null`` on the first page.
* @param data The data for the request.
* @param data.dagId
* @param data.dagRunId
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index 0c69edfd97b..f1d64923f48 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -341,7 +341,8 @@ export const useDagRunServiceGetDagRun = <TData =
Common.DagRunServiceGetDagRunD
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
runs match).
* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
* on the first page.
* @param data The data for the request.
@@ -1105,9 +1106,10 @@ export const useTaskInstanceServiceGetMappedTaskInstance
= <TData = Common.TaskI
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
-* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
-* on the first page.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
task instances
+* match). ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is
+* ``null`` on the first page.
* @param data The data for the request.
* @param data.dagId
* @param data.dagRunId
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index ec1e3504f7d..1c0bb7d0033 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -341,7 +341,8 @@ export const useDagRunServiceGetDagRunSuspense = <TData =
Common.DagRunServiceGe
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
runs match).
* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
* on the first page.
* @param data The data for the request.
@@ -1105,9 +1106,10 @@ export const
useTaskInstanceServiceGetMappedTaskInstanceSuspense = <TData = Comm
* **Offset (default):** use `limit` and `offset` query parameters. Returns
`total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
-* When `cursor` is provided, `offset` is ignored and `total_entries` is not
returned.
-* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
-* on the first page.
+* When `cursor` is provided, `offset` is ignored and `total_entries` is capped
at
+* `total_entries_limit` (a value equal to that limit means at least that many
task instances
+* match). ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is
+* ``null`` on the first page.
* @param data The data for the request.
* @param data.dagId
* @param data.dagRunId
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 1a1f29b0951..d9afa239aa2 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -3640,7 +3640,19 @@ export const $DAGRunCollectionResponse = {
}
],
title: 'Total Entries',
- description: 'Total number of matching items. Populated for offset
pagination, ``null`` when using cursor pagination.'
+ description: 'Number of matching items. For offset pagination this
is the exact total. For cursor pagination it is capped at
``total_entries_limit``; a value equal to that limit means at least that many
items match.'
+ },
+ total_entries_limit: {
+ anyOf: [
+ {
+ type: 'integer'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Total Entries Limit',
+ description: 'Cap applied to ``total_entries`` under cursor
pagination. ``null`` for offset pagination, where ``total_entries`` is exact.'
},
next_cursor: {
anyOf: [
@@ -6587,7 +6599,19 @@ export const $TaskInstanceCollectionResponse = {
}
],
title: 'Total Entries',
- description: 'Total number of matching items. Populated for offset
pagination, ``null`` when using cursor pagination.'
+ description: 'Number of matching items. For offset pagination this
is the exact total. For cursor pagination it is capped at
``total_entries_limit``; a value equal to that limit means at least that many
items match.'
+ },
+ total_entries_limit: {
+ anyOf: [
+ {
+ type: 'integer'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Total Entries Limit',
+ description: 'Cap applied to ``total_entries`` under cursor
pagination. ``null`` for offset pagination, where ``total_entries`` is exact.'
},
next_cursor: {
anyOf: [
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index 0fc5c1c220f..53f5da2de09 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -1103,7 +1103,8 @@ export class DagRunService {
* **Offset (default):** use `limit` and `offset` query parameters.
Returns `total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
- * When `cursor` is provided, `offset` is ignored and `total_entries` is
not returned.
+ * When `cursor` is provided, `offset` is ignored and `total_entries` is
capped at
+ * `total_entries_limit` (a value equal to that limit means at least that
many runs match).
* ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
* on the first page.
* @param data The data for the request.
@@ -2700,9 +2701,10 @@ export class TaskInstanceService {
* **Offset (default):** use `limit` and `offset` query parameters.
Returns `total_entries`.
*
* **Cursor:** pass `cursor` (empty string for the first page, then
`next_cursor` from the response).
- * When `cursor` is provided, `offset` is ignored and `total_entries` is
not returned.
- * ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is ``null``
- * on the first page.
+ * When `cursor` is provided, `offset` is ignored and `total_entries` is
capped at
+ * `total_entries_limit` (a value equal to that limit means at least that
many task instances
+ * match). ``next_cursor`` is ``null`` when there are no more pages;
``previous_cursor`` is
+ * ``null`` on the first page.
* @param data The data for the request.
* @param data.dagId
* @param data.dagRunId
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 92b30f77199..cf8c86e997c 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -1002,9 +1002,13 @@ export type DAGRunClearBody = {
export type DAGRunCollectionResponse = {
dag_runs: Array<DAGRunResponse>;
/**
- * Total number of matching items. Populated for offset pagination,
``null`` when using cursor pagination.
+ * Number of matching items. For offset pagination this is the exact
total. For cursor pagination it is capped at ``total_entries_limit``; a value
equal to that limit means at least that many items match.
*/
total_entries?: number | null;
+ /**
+ * Cap applied to ``total_entries`` under cursor pagination. ``null`` for
offset pagination, where ``total_entries`` is exact.
+ */
+ total_entries_limit?: number | null;
/**
* Token pointing to the next page. Populated for cursor pagination,
``null`` when using offset pagination or when there is no next page.
*/
@@ -1758,9 +1762,13 @@ export type TaskInletAssetReference = {
export type TaskInstanceCollectionResponse = {
task_instances: Array<TaskInstanceResponse>;
/**
- * Total number of matching items. Populated for offset pagination,
``null`` when using cursor pagination.
+ * Number of matching items. For offset pagination this is the exact
total. For cursor pagination it is capped at ``total_entries_limit``; a value
equal to that limit means at least that many items match.
*/
total_entries?: number | null;
+ /**
+ * Cap applied to ``total_entries`` under cursor pagination. ``null`` for
offset pagination, where ``total_entries`` is exact.
+ */
+ total_entries_limit?: number | null;
/**
* Token pointing to the next page. Populated for cursor pagination,
``null`` when using offset pagination or when there is no next page.
*/
diff --git
a/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.test.tsx
b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.test.tsx
index a4713410f36..aed86cfa8fb 100644
--- a/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.test.tsx
+++ b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.test.tsx
@@ -183,6 +183,42 @@ describe("DataTable", () => {
expect(screen.getByRole("heading")).toHaveTextContent("2 task");
});
+ it("renders row count heading for cursor pagination when total is provided",
() => {
+ render(
+ <DataTable
+ columns={columns}
+ data={data}
+ initialState={{ pagination, sorting: [] }}
+ modelName="task"
+ nextCursor="next"
+ onStateChange={onStateChange}
+ total={2}
+ totalEntriesLimit={50_000}
+ />,
+ { wrapper: ChakraWrapper },
+ );
+
+ expect(screen.getByRole("heading")).toHaveTextContent("2 task");
+ });
+
+ it("renders a capped row count heading when total reaches
totalEntriesLimit", () => {
+ render(
+ <DataTable
+ columns={columns}
+ data={data}
+ initialState={{ pagination, sorting: [] }}
+ modelName="task"
+ nextCursor="next"
+ onStateChange={onStateChange}
+ total={50_000}
+ totalEntriesLimit={50_000}
+ />,
+ { wrapper: ChakraWrapper },
+ );
+
+ expect(screen.getByRole("heading")).toHaveTextContent("50,000+ task");
+ });
+
it("does not render row count heading when showRowCountHeading is false", ()
=> {
render(
<DataTable
diff --git a/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
index 9adc95b6490..fbcec76a677 100644
--- a/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
+++ b/airflow-core/src/airflow/ui/src/components/DataTable/DataTable.tsx
@@ -63,6 +63,7 @@ type DataTableProps<TData> = {
readonly showRowCountHeading?: boolean;
readonly skeletonCount?: number;
readonly total?: number;
+ readonly totalEntriesLimit?: number;
};
const defaultGetRowCanExpand = () => false;
@@ -88,10 +89,11 @@ export const DataTable = <TData,>({
showRowCountHeading = true,
skeletonCount = 10,
total = 0,
+ totalEntriesLimit,
}: DataTableProps<TData>) => {
"use no memo"; // remove if https://github.com/TanStack/table/issues/5567 is
resolved
- const { t: translate } = useTranslation(["common"]);
+ const { i18n, t: translate } = useTranslation(["common"]);
const ref = useRef<{ tableRef: TanStackTable<TData> | undefined }>({
tableRef: undefined,
});
@@ -169,13 +171,17 @@ export const DataTable = <TData,>({
[modelName, translate],
);
const showRowCount = Boolean(
- showRowCountHeading && !hasCursorPagination && !Boolean(isLoading) &&
!Boolean(isFetching) && total > 0,
+ showRowCountHeading && !Boolean(isLoading) && !Boolean(isFetching) &&
total > 0,
);
+ // Cursor pagination reports the total capped at totalEntriesLimit, so a
total that reaches the
+ // cap means "at least this many" and is rendered as "N+".
+ const isCapped = totalEntriesLimit !== undefined && total >=
totalEntriesLimit;
+ const totalLabel = `${total.toLocaleString(i18n.language)}${isCapped ? "+" :
""}`;
const noRowsModelName = translateModelName(0);
const rowCountHeading = showRowCount ? (
<Heading py={3} size="md">
- {`${total} ${translateModelName(total)}`}
+ {`${totalLabel} ${translateModelName(total)}`}
</Heading>
) : undefined;
diff --git
a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.test.tsx
b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.test.tsx
index 824121bb042..4e5a88f66a8 100644
--- a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.test.tsx
@@ -48,6 +48,7 @@ vi.mock("openapi/queries", () => ({
vi.mock("react-i18next", () => ({
useTranslation: () => ({
+ i18n: { language: "en" },
// eslint-disable-next-line id-length
t: (key: string, options?: { id?: number }) =>
key === "components:backfill.viewSlots"
diff --git a/airflow-core/src/airflow/ui/src/pages/DagRuns/DagRuns.tsx
b/airflow-core/src/airflow/ui/src/pages/DagRuns/DagRuns.tsx
index 26a5fd0ae93..eb868302896 100644
--- a/airflow-core/src/airflow/ui/src/pages/DagRuns/DagRuns.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/DagRuns/DagRuns.tsx
@@ -387,6 +387,8 @@ export const DagRuns = () => {
nextCursor={nextCursor}
onStateChange={setTableURLState}
previousCursor={previousCursor}
+ total={data?.total_entries ?? 0}
+ totalEntriesLimit={data?.total_entries_limit ?? undefined}
/>
<ActionBar.Root closeOnInteractOutside={false}
open={Boolean(selectedRows.size)}>
<ActionBar.Content>
diff --git
a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
index 7f6a899adf6..a67e964d15d 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx
@@ -427,6 +427,8 @@ export const TaskInstances = () => {
nextCursor={nextCursor}
onStateChange={setTableURLState}
previousCursor={previousCursor}
+ total={data?.total_entries ?? 0}
+ totalEntriesLimit={data?.total_entries_limit ?? undefined}
/>
<ActionBar.Root closeOnInteractOutside={false}
open={Boolean(selectedRows.size)}>
<ActionBar.Content>
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index 974ebf82d3e..d574fc58637 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -678,7 +678,8 @@ class TestGetDagRuns:
body = response.json()
assert body["next_cursor"] is not None
assert body["previous_cursor"] is None
- assert body["total_entries"] is None
+ assert body["total_entries"] == 4
+ assert body["total_entries_limit"] == 50_000
assert len(body["dag_runs"]) == 2
response2 = test_client.get(
@@ -688,7 +689,8 @@ class TestGetDagRuns:
assert response2.status_code == 200, response2.json()
body2 = response2.json()
assert body2["previous_cursor"] is not None
- assert body2["total_entries"] is None
+ assert body2["total_entries"] == 4
+ assert body2["total_entries_limit"] == 50_000
assert len(body2["dag_runs"]) == 2
first_page_ids = {(r["dag_id"], r["dag_run_id"]) for r in
body["dag_runs"]}
second_page_ids = {(r["dag_id"], r["dag_run_id"]) for r in
body2["dag_runs"]}
@@ -700,14 +702,15 @@ class TestGetDagRuns:
)
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
def test_cursor_pagination_returns_cursor_response(self, test_client,
order_by):
- """When cursor param is provided, response has cursor fields and no
total_entries."""
+ """When cursor param is provided, response has cursor fields and a
bounded total_entries."""
response1 = test_client.get(
"/dags/~/dagRuns",
params={"limit": 2, "order_by": order_by, "cursor": ""},
)
assert response1.status_code == 200
body1 = response1.json()
- assert body1["total_entries"] is None
+ assert body1["total_entries"] == 4
+ assert body1["total_entries_limit"] == 50_000
assert len(body1["dag_runs"]) == 2
next_cursor = body1["next_cursor"]
assert next_cursor is not None
@@ -721,7 +724,8 @@ class TestGetDagRuns:
body2 = response2.json()
assert body2["next_cursor"] is None
assert body2["previous_cursor"] is not None
- assert body2["total_entries"] is None
+ assert body2["total_entries"] == 4
+ assert body2["total_entries_limit"] == 50_000
@pytest.mark.parametrize(
"order_by",
@@ -744,7 +748,8 @@ class TestGetDagRuns:
)
assert response.status_code == 200, response.json()
body = response.json()
- assert body["total_entries"] is None
+ assert body["total_entries"] == total_runs
+ assert body["total_entries_limit"] == 50_000
forward_pages.append(body)
forward_ids.extend((r["dag_id"], r["dag_run_id"]) for r in
body["dag_runs"])
@@ -779,6 +784,20 @@ class TestGetDagRuns:
all_backward = backward_ids + [(r["dag_id"], r["dag_run_id"]) for r in
forward_pages[-1]["dag_runs"]]
assert all_backward == forward_ids
+ @mock.patch("airflow.api_fastapi.common.db.common.EXACT_COUNT_LIMIT", 2)
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_cursor_pagination_total_entries_capped(self, test_client):
+ """With more matching runs than the cap, total_entries is reported as
the cap."""
+ response = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": 1, "order_by": "id", "cursor": ""},
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ # 4 runs match but the count stops scanning at the cap.
+ assert body["total_entries"] == 2
+ assert body["total_entries_limit"] == 2
+
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
def test_cursor_pagination_invalid_token(self, test_client):
response = test_client.get(
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
index 6dc6ecac582..0efc3e003e0 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
@@ -2081,11 +2081,12 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
body = response.json()
assert body["next_cursor"] is not None
assert body["previous_cursor"] is None
- assert body["total_entries"] is None
+ assert body["total_entries"] == 5
+ assert body["total_entries_limit"] == 50_000
assert len(body["task_instances"]) == 3
def test_cursor_pagination_returns_cursor_response(self, test_client,
session):
- """When cursor param is provided, response has cursor fields and no
total_entries."""
+ """When cursor param is provided, response has cursor fields and a
bounded total_entries."""
dag_id = "example_python_operator"
self.create_task_instances(
session,
@@ -2101,7 +2102,8 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
)
assert response1.status_code == 200
body1 = response1.json()
- assert body1["total_entries"] is None
+ assert body1["total_entries"] == 5
+ assert body1["total_entries_limit"] == 50_000
assert len(body1["task_instances"]) == 3
next_cursor = body1["next_cursor"]
assert next_cursor is not None
@@ -2115,7 +2117,8 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
body2 = response2.json()
assert body2["next_cursor"] is None
assert body2["previous_cursor"] is not None
- assert body2["total_entries"] is None
+ assert body2["total_entries"] == 5
+ assert body2["total_entries_limit"] == 50_000
def test_cursor_pagination_forward_and_backward_consistency(self,
test_client, session):
"""Walk all pages forward via next_cursor, then backward via
previous_cursor, and compare."""
@@ -2142,7 +2145,8 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
)
assert response.status_code == 200, response.json()
body = response.json()
- assert body["total_entries"] is None
+ assert body["total_entries"] == total_tis
+ assert body["total_entries_limit"] == 50_000
forward_pages.append(body)
forward_ids.extend(ti["id"] for ti in body["task_instances"])
@@ -4896,6 +4900,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
}
@@ -5173,6 +5178,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
},
@@ -5312,6 +5318,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
}
@@ -5376,6 +5383,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
}
@@ -5472,6 +5480,7 @@ class TestPatchTaskInstance(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
}
@@ -5752,6 +5761,7 @@ class
TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
}
@@ -6041,6 +6051,7 @@ class
TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
}
],
"total_entries": 1,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
},
@@ -6124,6 +6135,7 @@ class
TestPatchTaskInstanceDryRun(TestTaskInstanceEndpoint):
assert response.json() == {
"task_instances": [],
"total_entries": 0,
+ "total_entries_limit": None,
"next_cursor": None,
"previous_cursor": None,
}
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index ca3c63d6168..3e9681b4c18 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -2368,10 +2368,17 @@ class DAGRunCollectionResponse(BaseModel):
total_entries: Annotated[
int | None,
Field(
- description="Total number of matching items. Populated for offset
pagination, ``null`` when using cursor pagination.",
+ description="Number of matching items. For offset pagination this
is the exact total. For cursor pagination it is capped at
``total_entries_limit``; a value equal to that limit means at least that many
items match.",
title="Total Entries",
),
] = None
+ total_entries_limit: Annotated[
+ int | None,
+ Field(
+ description="Cap applied to ``total_entries`` under cursor
pagination. ``null`` for offset pagination, where ``total_entries`` is exact.",
+ title="Total Entries Limit",
+ ),
+ ] = None
next_cursor: Annotated[
str | None,
Field(
@@ -2491,10 +2498,17 @@ class TaskInstanceCollectionResponse(BaseModel):
total_entries: Annotated[
int | None,
Field(
- description="Total number of matching items. Populated for offset
pagination, ``null`` when using cursor pagination.",
+ description="Number of matching items. For offset pagination this
is the exact total. For cursor pagination it is capped at
``total_entries_limit``; a value equal to that limit means at least that many
items match.",
title="Total Entries",
),
] = None
+ total_entries_limit: Annotated[
+ int | None,
+ Field(
+ description="Cap applied to ``total_entries`` under cursor
pagination. ``null`` for offset pagination, where ``total_entries`` is exact.",
+ title="Total Entries Limit",
+ ),
+ ] = None
next_cursor: Annotated[
str | None,
Field(
