Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -277,18 +277,40 @@ def _add_cursor_limit_to_query(
order_by: str | None,
ascending: bool | None,
) -> SqlAlchemyQuery:
"""Helper function to deal with cursor/limit pagination args."""
"""Helper function to deal with cursor/limit/ordering pagination args for a RunsTable query.

Cursor pagination is only supported when sorting by ``RunsTable.c.id`` (the default). If
``order_by`` names a different column and ``cursor`` is also provided, this method raises.
"""
sorting_column: db.Column = getattr(RunsTable.c, order_by) if order_by else RunsTable.c.id
if cursor:
cursor_query = db_select([RunsTable.c.id]).where(RunsTable.c.run_id == cursor)
if ascending:
query = query.where(RunsTable.c.id > db_scalar_subquery(cursor_query))
if sorting_column is RunsTable.c.id:
# `cursor_query` and `query.where(RunsTable.c.id ....)` both assume that `id` is
# the sorting_column.
cursor_query = db_select([RunsTable.c.id]).where(RunsTable.c.run_id == cursor)
if ascending:
query = query.where(RunsTable.c.id > db_scalar_subquery(cursor_query))
else:
query = query.where(RunsTable.c.id < db_scalar_subquery(cursor_query))
elif sorting_column.unique:
# Single-column cursor pagination requires a unique sort column to identify the last
# seen row unambiguously. This branch means the column is unique but is not `id`.
# The only other unique column today is `run_id`, which is a random uuid4 and not a
# useful sort key. Stable, multi-column cursor pagination over a composite index
# is possible in general but not currently implemented.
check.not_implemented(
f"Cursor pagination for this RunsTable unique column: {order_by}"
)
else:
query = query.where(RunsTable.c.id < db_scalar_subquery(cursor_query))
check.param_invariant(
False,
"order_by",
f"Cursor pagination requires a unique sort column, but RunsTable.c.{order_by} is not unique.",
)
Comment thread
jmoldow marked this conversation as resolved.

if limit:
query = query.limit(limit)

sorting_column = getattr(RunsTable.c, order_by) if order_by else RunsTable.c.id
direction = db.asc if ascending else db.desc
query = query.order_by(direction(sorting_column))

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
from dagster._serdes import serialize_pp
from dagster._time import create_datetime, datetime_from_timestamp
from dagster_shared import seven
from dagster_shared.check.functions import NotImplementedCheckError, ParameterCheckError
from dagster_test.utils.data_factory import dagster_run as create_dagster_run

win_py36 = seven.IS_WINDOWS and sys.version_info[0] == 3 and sys.version_info[1] == 6
Expand Down Expand Up @@ -940,6 +941,24 @@ def test_fetch_records_by_update_timestamp(self, storage, instance):
)
] == [two]

def test_cursor_pagination_with_non_unique_order_by_raises(self, storage):
"""Cursor pagination on a non-unique column is undefined and must raise ParameterCheckError."""
assert storage
self._skip_in_memory(storage)

with pytest.raises(ParameterCheckError, match="requires a unique sort column"):
storage.get_run_records(cursor=str(uuid4()), order_by="status")

def test_cursor_pagination_with_unique_non_id_order_by_raises(self, storage):
"""Cursor pagination on a unique non-id column is not yet implemented."""
assert storage
self._skip_in_memory(storage)

with pytest.raises(
NotImplementedCheckError, match="Cursor pagination for this RunsTable unique column"
):
storage.get_run_records(cursor=str(uuid4()), order_by="run_id")

def test_fetch_records_by_create_timestamp(self, storage):
assert storage
self._skip_in_memory(storage)
Expand Down