This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new c183d07374a Fix cursor pagination dropping rows when sorting by a
nullable column (#68869) (#70739)
c183d07374a is described below
commit c183d07374a6d3a8b859cdfa8738175bfec9b2a3
Author: Rahul Vats <[email protected]>
AuthorDate: Thu Jul 30 21:50:21 2026 +0530
Fix cursor pagination dropping rows when sorting by a nullable column
(#68869) (#70739)
* Fix cursor pagination dropping rows when sorting by a nullable column
Cursor (keyset) paginated REST list endpoints silently dropped rows when the
sort column was nullable. The keyset predicate and the generated ORDER BY
disagreed on where NULLs sort, so once a page boundary fell on the
NULL/non-NULL
edge every row on one side of it was skipped with no error. This is silent
data
loss from the public API under a common sort such as start_date.
Cross-dialect NULLS FIRST/LAST is not portable (unsupported on MySQL and old
SQLite), so the cursor path now pins NULL placement with a portable
CASE-based
null-rank key shared by both the keyset ORDER BY and the keyset predicate,
so the
two can no longer disagree on any backend. The rank follows the column's
sort
direction, so NULLs order as the largest value (last when ascending, first
when
descending), matching PostgreSQL's default. PostgreSQL result order is
therefore
unchanged; SQLite and MySQL NULL ordering shifts to align with it. The
cursor
token format is unchanged (the rank is derived, not encoded) and offset
pagination is untouched.
* Add newsfragment for nullable-column cursor pagination fix
* Preserve index usage for nullable-column cursor pagination
The keyset ORDER BY ranked NULLs with a portable CASE expression so NULL
placement was uniform across backends. A computed ORDER BY key cannot use a
B-tree index, so large lists fall back to a full sort. Match each backend's
native NULL placement in the keyset predicate instead and keep the ORDER BY
a
bare column, preserving the index; the fix still returns every row exactly
once.
EXPLAIN on Postgres confirms an Index Only Scan rather than a Sort.
(cherry picked from commit 3b13e593ac697c5cb3dd6ded9c63e0f5c24a547a)
Co-authored-by: Steve Ahn <[email protected]>
---
.../src/airflow/api_fastapi/common/cursors.py | 66 ++++++-----
.../api_fastapi/core_api/routes/public/dag_run.py | 7 +-
.../core_api/routes/public/task_instances.py | 11 +-
.../tests/unit/api_fastapi/common/test_cursors.py | 123 +++++++++++++++++++--
.../core_api/routes/public/test_dag_run.py | 34 ++++++
.../core_api/routes/public/test_task_instances.py | 101 +++++++++++++++++
6 files changed, 299 insertions(+), 43 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/cursors.py
b/airflow-core/src/airflow/api_fastapi/common/cursors.py
index 2eb5569f890..c775e72ab70 100644
--- a/airflow-core/src/airflow/api_fastapi/common/cursors.py
+++ b/airflow-core/src/airflow/api_fastapi/common/cursors.py
@@ -28,7 +28,7 @@ from typing import Any
import msgspec
from fastapi import HTTPException, status
-from sqlalchemy import and_, or_
+from sqlalchemy import and_, false, or_, true
from sqlalchemy.sql import Select
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.sql.sqltypes import Uuid
@@ -43,55 +43,57 @@ def _b64url_decode_padded(token: str) -> bytes:
return base64.urlsafe_b64decode(token.encode("ascii"))
-def _nonstrict_bound(col: ColumnElement, value: Any, is_desc: bool) ->
ColumnElement[bool]:
+def _dialect_nulls_last(is_desc: bool, dialect: str) -> bool:
"""
- Inclusive range edge on the leading column at each nesting level (``>=`` /
``<=``).
+ Where a plain ``ORDER BY col`` puts NULLs on this backend.
- When *value* is ``None`` the column is nullable and the cursor sits at a
- NULL boundary. ``col IS NULL`` is used instead of ``col >= NULL`` (which
- SQLAlchemy rejects and SQL evaluates as UNKNOWN).
+ PostgreSQL sorts NULLs last for ASC, first for DESC; MySQL and SQLite treat
+ NULL as the lowest value, so NULLs come first for ASC and last for DESC.
"""
- if value is None:
- return col.is_(None)
- return col <= value if is_desc else col >= value
+ return (not is_desc) if dialect == "postgresql" else is_desc
-def _strict_bound(col: ColumnElement, value: Any, is_desc: bool) ->
ColumnElement[bool]:
+def _bounds(
+ col: ColumnElement, value: Any, is_desc: bool, dialect: str
+) -> tuple[ColumnElement[bool], ColumnElement[bool]]:
"""
- Strict inequality for ``or_`` branches (``<`` / ``>``).
-
- When *value* is ``None`` the cursor is at a NULL boundary. The only rows
- that can be "strictly after" a NULL are non-NULL rows (regardless of
- whether the database sorts NULLs first or last), so ``col IS NOT NULL`` is
- used. When the surrounding ``_nonstrict_bound`` already constrains
- ``col IS NULL``, this branch evaluates to FALSE and the inner keyset
- predicate takes over — which is the correct behaviour.
+ ``(non_strict, strict)`` keyset bounds matching the backend's native NULL
placement.
+
+ A NULL *value* means the cursor sits in the NULL block; otherwise a
trailing
+ NULL block (when NULLs sort last) is also admitted after a non-NULL cursor.
+ The ``col IS NULL`` terms are vacuous for non-nullable columns.
"""
+ nulls_last = _dialect_nulls_last(is_desc, dialect)
if value is None:
- return col.is_not(None)
- return col < value if is_desc else col > value
+ if nulls_last:
+ return col.is_(None), false()
+ return true(), col.is_not(None)
+ ge = col <= value if is_desc else col >= value
+ gt = col < value if is_desc else col > value
+ if nulls_last:
+ return or_(ge, col.is_(None)), or_(gt, col.is_(None))
+ return ge, gt
def _nested_keyset_predicate(
- resolved: list[tuple[str, ColumnElement, bool]], values: list[Any]
+ resolved: list[tuple[str, ColumnElement, bool]], values: list[Any],
dialect: str
) -> ColumnElement[bool]:
"""
Keyset predicate for rows strictly after the cursor in ``ORDER BY`` order.
Uses nested ``and_(non-strict, or_(strict, ...))`` so leading sort keys use
inclusive range bounds and inner branches use strict inequalities—friendly
- for composite index range scans. Logically equivalent to an OR-of-prefix-
- equalities formulation.
+ for composite index range scans. NULL placement follows each backend's
+ native ordering for a plain ``ORDER BY`` (see :func:`_bounds`), so the
+ ``ORDER BY`` stays a bare column and can still use an index.
"""
n = len(resolved)
_, col, is_desc = resolved[n - 1]
- inner: ColumnElement[bool] = _strict_bound(col, values[n - 1], is_desc)
+ _, inner = _bounds(col, values[n - 1], is_desc, dialect)
for i in range(n - 2, -1, -1):
_, col_i, is_desc_i = resolved[i]
- inner = and_(
- _nonstrict_bound(col_i, values[i], is_desc_i),
- or_(_strict_bound(col_i, values[i], is_desc_i), inner),
- )
+ non_strict, strict = _bounds(col_i, values[i], is_desc_i, dialect)
+ inner = and_(non_strict, or_(strict, inner))
return inner
@@ -163,7 +165,7 @@ def decode_cursor(token: str) -> list[Any]:
def apply_cursor_filter(
- statement: Select, token: str, sort_param: SortParam, *, is_backward: bool
= False
+ statement: Select, token: str, sort_param: SortParam, dialect: str, *,
is_backward: bool = False
) -> Select:
"""
Apply a keyset pagination WHERE clause from a cursor token.
@@ -173,6 +175,10 @@ def apply_cursor_filter(
flipped so the predicate selects rows strictly *before* the cursor in the
original sort order. The caller is responsible for reversing the ORDER BY
and the final result list when using a backward cursor.
+
+ *dialect* is the backend dialect name
(``session.get_bind().dialect.name``);
+ the predicate matches that backend's native NULL placement, so the keyset
+ ``ORDER BY`` stays a bare column and keeps using indexes.
"""
raw_values = decode_cursor(token)
@@ -185,4 +191,4 @@ def apply_cursor_filter(
if is_backward:
resolved = [(name, col, not is_desc) for name, col, is_desc in
resolved]
- return statement.where(_nested_keyset_predicate(resolved, parsed_values))
+ return statement.where(_nested_keyset_predicate(resolved, parsed_values,
dialect))
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 84e2f77f8fa..7b03c7b3ece 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
@@ -663,14 +663,17 @@ def get_dag_runs(
"int", limit.value
) # LimitFilter value is guaranteed to be set to the default value of
QueryLimit
cursor_limit = LimitFilter().set_value(page_limit + 1)
- dag_run_select = apply_filters_to_select(statement=query,
filters=[*filters, order_by, cursor_limit])
+ dag_run_select = apply_filters_to_select(statement=query,
filters=[*filters, cursor_limit])
+ dag_run_select = order_by.to_orm(dag_run_select)
is_backward = False
if cursor:
token, is_backward = parse_cursor(cursor)
if is_backward:
dag_run_select = order_by.to_orm(dag_run_select, reversed=True)
- dag_run_select = apply_cursor_filter(dag_run_select, token,
order_by, is_backward=is_backward)
+ dag_run_select = apply_cursor_filter(
+ dag_run_select, token, order_by,
session.get_bind().dialect.name, is_backward=is_backward
+ )
fetched = list(session.scalars(dag_run_select))
has_more = len(fetched) > page_limit
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 fc28c377fe6..7305e4e89be 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
@@ -607,9 +607,8 @@ def get_task_instances(
"int", limit.value
) # LimitFilter value is guaranteed to be set to the default value of
QueryLimit
cursor_limit = LimitFilter().set_value(page_limit + 1)
- task_instance_select = apply_filters_to_select(
- statement=query, filters=[*filters, order_by, cursor_limit]
- )
+ task_instance_select = apply_filters_to_select(statement=query,
filters=[*filters, cursor_limit])
+ task_instance_select = order_by.to_orm(task_instance_select)
is_backward = False
if cursor:
@@ -617,7 +616,11 @@ def get_task_instances(
if is_backward:
task_instance_select = order_by.to_orm(task_instance_select,
reversed=True)
task_instance_select = apply_cursor_filter(
- task_instance_select, token, order_by, is_backward=is_backward
+ task_instance_select,
+ token,
+ order_by,
+ session.get_bind().dialect.name,
+ is_backward=is_backward,
)
fetched = list(session.scalars(task_instance_select))
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_cursors.py
b/airflow-core/tests/unit/api_fastapi/common/test_cursors.py
index 8bf8adbba36..f101efebc9d 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_cursors.py
+++ b/airflow-core/tests/unit/api_fastapi/common/test_cursors.py
@@ -26,7 +26,7 @@ from unittest.mock import MagicMock
import msgspec
import pytest
from fastapi import HTTPException
-from sqlalchemy import select
+from sqlalchemy import Column, String, select
from airflow.api_fastapi.common.cursors import apply_cursor_filter,
decode_cursor, encode_cursor
from airflow.api_fastapi.common.parameters import SortParam
@@ -126,7 +126,7 @@ class TestCursorPagination:
token = _msgpack_cursor_token(["only-one-value"])
with pytest.raises(HTTPException, match="does not match"):
- apply_cursor_filter(select(TaskInstance), token, sp)
+ apply_cursor_filter(select(TaskInstance), token, sp, "sqlite")
def test_apply_cursor_filter_ascending(self):
sp = self._make_sort_param_with_resolved_columns(["start_date"])
@@ -136,7 +136,7 @@ class TestCursorPagination:
]
token = _msgpack_cursor_token(values)
- stmt = apply_cursor_filter(select(TaskInstance), token, sp)
+ stmt = apply_cursor_filter(select(TaskInstance), token, sp, "sqlite")
sql = str(stmt)
assert ">" in sql
@@ -148,7 +148,7 @@ class TestCursorPagination:
]
token = _msgpack_cursor_token(values)
- stmt = apply_cursor_filter(select(TaskInstance), token, sp)
+ stmt = apply_cursor_filter(select(TaskInstance), token, sp, "sqlite")
sql = str(stmt)
assert "<" in sql
@@ -193,8 +193,117 @@ class TestCursorPagination:
sp.set_value(["_rendered_map_index", "map_index"])
token = _msgpack_cursor_token([None, 49,
"019462ab-1234-5678-9abc-def012345678"])
- # Should not raise ArgumentError from SQLAlchemy.
- stmt = apply_cursor_filter(select(TaskInstance), token, sp)
+ # Should not raise ArgumentError from SQLAlchemy; the NULL boundary is
+ # expressed with IS [NOT] NULL rather than a comparison against NULL.
+ stmt = apply_cursor_filter(select(TaskInstance), token, sp, "sqlite")
sql = str(stmt)
- assert "IS NULL" in sql
assert "IS NOT NULL" in sql
+
+
+class TestKeysetPaginationNullableColumn:
+ """End-to-end: cursor pagination over a nullable sort column returns every
row exactly once.
+
+ When the keyset predicate and the ORDER BY disagree on where NULLs fall,
one side
+ of the NULL/non-NULL boundary is silently dropped. The predicate matches
each
+ backend's native NULL placement so the ORDER BY stays a bare (indexable)
column.
+ """
+
+ @staticmethod
+ def _seed_session(rows):
+ """Build an in-memory model with one nullable column and seed it with
``(id, val)`` rows."""
+ from sqlalchemy import Integer, create_engine
+ from sqlalchemy.orm import Session, declarative_base
+
+ base = declarative_base()
+
+ class Item(base):
+ __tablename__ = "keyset_items"
+ id = Column(Integer, primary_key=True)
+ val = Column(String, nullable=True)
+
+ engine = create_engine("sqlite://")
+ base.metadata.create_all(engine)
+ session = Session(engine)
+ session.add_all([Item(id=i, val=v) for i, v in rows])
+ session.commit()
+ return Item, session
+
+ @staticmethod
+ def _walk_forward(session, model, sort, page_size):
+ collected: list[int] = []
+ token = None
+ for _ in range(50): # guard against an infinite paging loop
+ stmt = sort.to_orm(select(model)).limit(page_size)
+ if token is not None:
+ stmt = apply_cursor_filter(stmt, token, sort, "sqlite")
+ rows = list(session.scalars(stmt))
+ if not rows:
+ break
+ collected.extend(r.id for r in rows)
+ token = encode_cursor(rows[-1], sort)
+ return collected
+
+ @pytest.mark.parametrize(
+ ("order_by", "expected_order"),
+ [
+ pytest.param(["val"], [1, 2, 3, 4, 5, 6],
id="ascending-nulls-first"),
+ pytest.param(["-val"], [6, 5, 4, 3, 2, 1],
id="descending-nulls-last"),
+ ],
+ )
+ def test_forward_pagination_returns_all_rows(self, order_by,
expected_order):
+ model, session = self._seed_session([(1, None), (2, None), (3, None),
(4, "a"), (5, "b"), (6, "c")])
+ try:
+ sort = SortParam(["val"], model)
+ sort.set_value(order_by)
+ collected = self._walk_forward(session, model, sort, page_size=2)
+ finally:
+ session.close()
+
+ assert sorted(collected) == [1, 2, 3, 4, 5, 6], f"rows
dropped/duplicated: {collected}"
+ assert len(collected) == len(set(collected)), f"rows duplicated:
{collected}"
+ assert collected == expected_order
+
+ def test_multiple_nullable_columns_no_rows_dropped(self):
+ """Two nullable sort columns: every NULL/non-NULL combination paged
exactly once."""
+ from sqlalchemy import Integer, create_engine
+ from sqlalchemy.orm import Session, declarative_base
+
+ base = declarative_base()
+
+ class Pair(base):
+ __tablename__ = "keyset_pairs"
+ id = Column(Integer, primary_key=True)
+ a = Column(String, nullable=True)
+ b = Column(String, nullable=True)
+
+ engine = create_engine("sqlite://")
+ base.metadata.create_all(engine)
+ rows = [
+ (1, None, None),
+ (2, None, "y"),
+ (3, "p", None),
+ (4, "p", "y"),
+ (5, "q", None),
+ (6, "q", "z"),
+ ]
+ with Session(engine) as session:
+ session.add_all([Pair(id=i, a=a, b=b) for i, a, b in rows])
+ session.commit()
+ sort = SortParam(["a", "b"], Pair)
+ sort.set_value(["a", "b"])
+ collected = self._walk_forward(session, Pair, sort, page_size=2)
+
+ assert sorted(collected) == [1, 2, 3, 4, 5, 6], f"rows
dropped/duplicated: {collected}"
+ assert len(collected) == len(set(collected)), f"rows duplicated:
{collected}"
+
+ def test_nullable_column_without_nulls_unaffected(self):
+ """The common case (a nullable column that happens to hold no NULLs)
still pages correctly."""
+ model, session = self._seed_session([(1, "a"), (2, "b"), (3, "c"), (4,
"d"), (5, "e")])
+ try:
+ sort = SortParam(["val"], model)
+ sort.set_value(["val"])
+ collected = self._walk_forward(session, model, sort, page_size=2)
+ finally:
+ session.close()
+
+ assert collected == [1, 2, 3, 4, 5]
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 66bbadf2c85..d8b7eeae5e1 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
@@ -690,6 +690,40 @@ class TestGetDagRuns:
)
assert response.status_code == 400
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_cursor_pagination_nullable_sort_column_returns_all_rows(self,
test_client, session):
+ """Cursor pagination sorted by a nullable column must not silently
drop rows.
+
+ With a NULL in the sort column, the keyset predicate and the ORDER BY
can disagree
+ on NULL placement and drop every row on one side of the NULL/non-NULL
boundary.
+ """
+ # Null out one run's start_date so the NULL/non-NULL boundary is
crossed mid-walk.
+ run = session.scalar(select(DagRun).where(DagRun.run_id ==
DAG1_RUN1_ID))
+ run.start_date = None
+ session.commit()
+
+ full = test_client.get("/dags/~/dagRuns", params={"limit": 100})
+ assert full.status_code == 200, full.json()
+ full_ids = {(r["dag_id"], r["dag_run_id"]) for r in
full.json()["dag_runs"]}
+ assert len(full_ids) == 4
+
+ collected: list[tuple[str, str]] = []
+ cursor_token: str | None = ""
+ for _ in range(20):
+ resp = test_client.get(
+ "/dags/~/dagRuns",
+ params={"limit": 1, "order_by": "start_date", "cursor":
cursor_token},
+ )
+ assert resp.status_code == 200, resp.json()
+ body = resp.json()
+ collected.extend((r["dag_id"], r["dag_run_id"]) for r in
body["dag_runs"])
+ cursor_token = body.get("next_cursor")
+ if cursor_token is None:
+ break
+
+ assert len(collected) == len(set(collected)), "cursor pages overlapped"
+ assert set(collected) == full_ids, "cursor pagination dropped rows
across the NULL boundary"
+
@pytest.mark.parametrize(
("dag_id", "query_params", "expected_dag_id_list"),
[
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 266ea938104..ae3a8388fea 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
@@ -2154,6 +2154,107 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
assert ids1.isdisjoint(ids2), "Pages must not overlap"
assert len(ids1) + len(ids2) == total_tis
+ def test_cursor_pagination_nullable_sort_column_returns_all_rows(self,
test_client, session):
+ """Cursor pagination sorted by a nullable column must not silently
drop rows.
+
+ With NULLs present, the keyset predicate and the ORDER BY can disagree
on NULL
+ placement, so every row on one side of the NULL/non-NULL boundary is
dropped
+ without error.
+ """
+ dag_id = "example_python_operator"
+ # Three TIs with NULL start_date and three with distinct values, so
both the NULL and
+ # the non-NULL block span more than one page and the boundary is
crossed mid-walk.
+ self.create_task_instances(
+ session,
+ task_instances=[
+ {"start_date": None},
+ {"start_date": None},
+ {"start_date": None},
+ {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=1)},
+ {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=2)},
+ {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=3)},
+ ],
+ dag_id=dag_id,
+ )
+
+ # Full set via offset pagination (returns everything).
+ full = test_client.get("/dags/~/dagRuns/~/taskInstances",
params={"limit": 100})
+ assert full.status_code == 200, full.json()
+ full_ids = {ti["id"] for ti in full.json()["task_instances"]}
+ assert len(full_ids) == 6
+
+ # Walk every page forward via cursor, sorted by the nullable column.
+ collected: list[str] = []
+ cursor_token: str | None = ""
+ for _ in range(20):
+ resp = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": 2, "order_by": ["start_date"], "cursor":
cursor_token},
+ )
+ assert resp.status_code == 200, resp.json()
+ body = resp.json()
+ collected.extend(ti["id"] for ti in body["task_instances"])
+ cursor_token = body.get("next_cursor")
+ if cursor_token is None:
+ break
+
+ assert len(collected) == len(set(collected)), "cursor pages overlapped"
+ assert set(collected) == full_ids, "cursor pagination dropped rows
across the NULL boundary"
+
+ def test_cursor_pagination_forward_backward_consistency_nullable(self,
test_client, session):
+ """Forward then backward walk over a nullable column must agree, NULLs
included.
+
+ Backward pagination flips the sort direction, re-deriving the keyset
bounds; this
+ guards that NULL placement stays consistent in both directions.
+ """
+ dag_id = "example_python_operator"
+ page_size = 3
+ # 3 NULL start_dates + 5 distinct values -> NULL block and non-NULL
block both span pages.
+ self.create_task_instances(
+ session,
+ task_instances=[{"start_date": None} for _ in range(3)]
+ + [{"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=(i +
1))} for i in range(5)],
+ dag_id=dag_id,
+ )
+
+ forward_ids: list[str] = []
+ forward_pages: list[dict] = []
+ cursor_token: str | None = ""
+ for _ in range(20):
+ response = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": page_size, "order_by": ["start_date"],
"cursor": cursor_token},
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ forward_pages.append(body)
+ forward_ids.extend(ti["id"] for ti in body["task_instances"])
+ cursor_token = body.get("next_cursor")
+ if cursor_token is None:
+ break
+
+ assert len(forward_ids) == 8
+ assert len(forward_ids) == len(set(forward_ids)), "Forward pages
should not overlap"
+ assert forward_pages[0]["previous_cursor"] is None
+
+ backward_ids: list[str] = []
+ cursor_token = forward_pages[-1]["previous_cursor"]
+ assert cursor_token is not None
+ for _ in range(20):
+ response = test_client.get(
+ "/dags/~/dagRuns/~/taskInstances",
+ params={"limit": page_size, "order_by": ["start_date"],
"cursor": cursor_token},
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ backward_ids = [ti["id"] for ti in body["task_instances"]] +
backward_ids
+ cursor_token = body.get("previous_cursor")
+ if cursor_token is None:
+ break
+
+ all_backward = backward_ids + [ti["id"] for ti in
forward_pages[-1]["task_instances"]]
+ assert all_backward == forward_ids, "Backward walk + last page must
match the forward walk exactly"
+
class TestGetTaskDependencies(TestTaskInstanceEndpoint):
def setup_method(self):
