fix(run-storage): validate cursor order_by column

## Summary & Motivation

Cursor pagination is invalid when the `order_by` doesn't match
the `cursor` and/or the cursor doesn't completely cover the
index. Validate that this doesn't happen, and raise if it ever
does happen.

`_add_cursor_limit_to_query` in `SqlRunStorage` previously
allowed semantically invalid cursor pagination with a non-`id`
`order_by` column. This would silently apply `id`-based cursor
filtering while sorting by the requested column — producing
incorrect, page-skipping results with no error.

This change adds explicit guards:

- If `cursor` is provided and `order_by` maps to a **non-unique
  column** (e.g. `status`), a `ParameterCheckError` is raised
  immediately, since cursor pagination on a non-unique column is
  undefined.
- If `cursor` is provided and `order_by` maps to a **unique non-`id`
  column** (e.g. `run_id`), a `NotImplementedCheckError` is raised,
  since stable single-column cursor pagination over an arbitrary unique
  key is not yet implemented.
- The existing `id`-based cursor path is unchanged.

## How I Tested These Changes

Two new test cases in `TestRunStorage` cover both error branches:
`test_cursor_pagination_with_non_unique_order_by_raises` and
`test_cursor_pagination_with_unique_non_id_order_by_raises`.

## Changelog

`DagsterInstance.get_run_records`, `RunStorage.get_runs`, and
`RunStorage.get_run_records`: passing `cursor` together with an
`order_by` column other than `id` now raises an explicit error
instead of silently returning wrong results.

Author: jmoldow

python_modules/dagster/dagster/_core/storage/runs/sql_run_storage.py

@@ -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,
+                    "sorting_column",
+                    f"Cursor pagination requires a unique sort column, but RunsTable.c.{order_by} is not unique.",
+                )
 
         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))
 

python_modules/dagster/dagster_tests/storage_tests/utils/run_storage.py

@@ -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
@@ -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)