feat: a Pager that cuts a keyset page and the cursor to the next

README.md

@@ -224,18 +224,38 @@ A `Cursor` is a resolved `OrderByClause` plus one boundary value per key,
 rendered as the OR-of-ANDs tuple comparison the seek method wants — not
 `OFFSET`, which re-reads every row it skips.
 
+A `Pager` runs the page loop around it. It knows the page size and the
+ordering, asks the query for one row past the page, and cuts what comes back
+into a `Page` — the rows, and the cursor to the next page if that extra row
+came back:
+
 ```rust
-// The page just fetched, turned into the cursor for the next one. Values come
-// off the last row, so no one has to know each key's Rust type.
-let cursor = Cursor::new(order_by.clone()).after_row(users.last().unwrap())?;
-let page_token = cursor.encode();
+let pager = Pager::new(Cursor::new(order_by.clone()), 50);
 
-// ... and on the next request:
-let cursor = Cursor::parse(&page_token)?;
 let mut query = QueryComposer::<Postgres>::new(LIST_USERS);
-query.bind_value(tenant_id).bind_value(50i64).with_cursor(cursor);
+query
+    .bind_value(tenant_id)
+    .bind_value(pager.limit()) // 51: one past the page
+    .push_order_by(order_by);
+if let Some(cursor) = cursor {
+    query.with_cursor(cursor); // where the previous page left off
+}
+
+let rows = query.build()?.fetch_all(&pool).await?;
+let page = pager.next_page(rows)?; // Page { rows: at most 50, cursor }
+
+// The cursor's values come off the page's last row, so no one has to know
+// each key's Rust type. `None` on the last page.
+let next_page_token = page.cursor.map(|cursor| cursor.encode());
+
+// ... and on the next request:
+let cursor = Some(Cursor::parse(&page_token)?);
 ```
 
+A page that exactly fills the size is the last one: the extra row is what says
+there is another. `Cursor::after_row` builds the same cursor by hand, from
+whichever row a caller picks.
+
 A key can be any column a `Value` holds, a UUID primary key included behind
 the `uuid` feature — it is read back off the row as a `Value::Uuid` and bound
 against the column on the next page.

crates/sqlx-query/src/clauses/mod.rs

@@ -1,7 +1,8 @@
 //! The clause types [`QueryComposer`](crate::QueryComposer)'s builder
 //! methods accept: [`WhereClause`] (`push_where`), [`OrderByClause`]
 //! (`push_order_by`), and [`Cursor`] (`with_cursor`, itself producing a
-//! `WhereClause`/`OrderByClause` pair for keyset pagination).
+//! `WhereClause`/`OrderByClause` pair for keyset pagination, with [`Pager`]
+//! cutting the fetched rows into a [`Page`] and the cursor to the next).
 
 mod order;
 mod pager;
@@ -11,5 +12,5 @@ mod pager;
 mod r#where;
 
 pub use order::{OrderByClause, OrderByClauseError, OrderDirection};
-pub use pager::{Cursor, CursorError};
+pub use pager::{Cursor, CursorError, Page, Pager};
 pub use r#where::WhereClause;

crates/sqlx-query/src/clauses/pager.rs

@@ -31,6 +31,11 @@
 //! decodes in turn and keeping whichever one `sqlx` itself accepts,
 //! rather than this crate re-deriving the same table from OIDs or type
 //! codes.
+//!
+//! [`Pager`] is the loop around it: a query fetches [`Pager::limit`] rows —
+//! one past the page — and [`Pager::next_page`] keeps the page and, when
+//! the extra row came back, calls [`Cursor::after_row`] on the last row
+//! kept. The result is a [`Page`]: the rows, and the cursor to the next.
 
 use serde::{Deserialize, Serialize};
 use sqlx::{ColumnIndex, Decode, Row, Type};
@@ -268,6 +273,102 @@ impl Cursor {
     }
 }
 
+/// Cuts pages out of what a keyset-paged query returns, and positions the
+/// cursor to the page after each.
+///
+/// Whether there is a next page is only knowable by asking for one row more
+/// than the page holds: [`limit`](Self::limit) is that many, for the query's
+/// `LIMIT`, and [`next_page`](Self::next_page) takes what came back, keeps
+/// the page, and reads the cursor off its last row if the extra row came
+/// back too. Both read the same `size`, so the page size is the only number
+/// a caller handles — the extra row never leaves this type.
+///
+/// A pager holds no position, only the ordering and the size, so one pager
+/// serves every page of a traversal.
+#[derive(Debug, Clone)]
+pub struct Pager {
+    cursor: Cursor,
+    size: usize,
+}
+
+impl Pager {
+    /// A pager for pages of `size` rows, ordered by `cursor`'s `order_by` —
+    /// typically [`Cursor::new`] over the same `order_by` the query is
+    /// sorted by. A position `cursor` already carries is replaced on each
+    /// page, so the cursor a request came in with works too.
+    ///
+    /// A `size` of 0 is taken as 1: a page with no rows could never say
+    /// where the next one starts.
+    #[must_use]
+    pub fn new(cursor: Cursor, size: usize) -> Self {
+        Pager {
+            cursor,
+            size: size.max(1),
+        }
+    }
+
+    /// How many rows the query should fetch: one past the page, so
+    /// [`next_page`](Self::next_page) can tell whether there is another.
+    /// Bind it to the query's `LIMIT`.
+    #[must_use]
+    pub fn limit(&self) -> i64 {
+        i64::try_from(self.size).map_or(i64::MAX, |size| size.saturating_add(1))
+    }
+
+    /// The page out of `rows` — what a query fetched with
+    /// [`limit`](Self::limit) returned, in its order.
+    ///
+    /// Keeps the page's `size` rows. If the extra row came back there is a
+    /// next page, and [`Page::cursor`] is positioned after the last row kept
+    /// (see [`Cursor::after_row`]); otherwise this is the last page and it is
+    /// `None`.
+    ///
+    /// # Errors
+    ///
+    /// As [`Cursor::after_row`], for the last row kept:
+    /// [`CursorError::RowColumnMissing`] or
+    /// [`CursorError::RowValueUndecodable`] for a key the row cannot supply.
+    pub fn next_page<R>(&self, mut rows: Vec<R>) -> Result<Page<R>, CursorError>
+    where
+        R: Row,
+        usize: ColumnIndex<R>,
+        for<'r> bool: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> i16: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> i32: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> i64: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> f32: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> f64: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> String: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> crate::value::TimestampRepr: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> Vec<u8>: Decode<'r, R::Database> + Type<R::Database>,
+        for<'r> crate::value::UuidRepr: Decode<'r, R::Database> + Type<R::Database>,
+    {
+        if rows.len() <= self.size {
+            return Ok(Page { rows, cursor: None });
+        }
+
+        // At least one row is left: `size` is never 0, and there were more.
+        rows.truncate(self.size);
+        let cursor = rows
+            .last()
+            .map(|last| self.cursor.clone().after_row(last))
+            .transpose()?;
+
+        Ok(Page { rows, cursor })
+    }
+}
+
+/// One page of rows, and where the next one starts — what
+/// [`Pager::next_page`] cuts out of a query's rows.
+#[derive(Debug, Clone)]
+pub struct Page<R> {
+    /// The page's rows, at most the pager's size.
+    pub rows: Vec<R>,
+    /// The cursor to the next page — [`encode`](Cursor::encode) it as the
+    /// `next_page_token` — or `None` on the last page.
+    pub cursor: Option<Cursor>,
+}
+
 /// `key`'s 1-based position in `keys` — its placeholder number, since
 /// each key's value is bound once and referenced by number wherever it
 /// recurs across the OR branches.
@@ -432,6 +533,16 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn a_pager_fetches_one_row_past_the_page() {
+        let cursor = Cursor::new(OrderByClause::default().asc("id"));
+
+        assert_eq!(Pager::new(cursor.clone(), 50).limit(), 51);
+        // An empty page could never say where the next one starts.
+        assert_eq!(Pager::new(cursor.clone(), 0).limit(), 2);
+        assert_eq!(Pager::new(cursor, usize::MAX).limit(), i64::MAX);
+    }
+
     #[test]
     fn checksum_mask_distinguishes_from_a_plain_crc32() {
         let order_by = OrderByClause::parse("rank desc").unwrap();

crates/sqlx-query/src/lib.rs

@@ -15,7 +15,8 @@ mod value;
 
 pub use argument::{Bindable, QueryArgument};
 pub use clauses::{
-    Cursor, CursorError, OrderByClause, OrderByClauseError, OrderDirection, WhereClause,
+    Cursor, CursorError, OrderByClause, OrderByClauseError, OrderDirection, Page, Pager,
+    WhereClause,
 };
 pub use composer::{Error, QueryComposer, QueryStatement};
 pub use dialect::{PlaceholderStyle, QueryDialect, QuerySyntax, Quoting};

crates/sqlx-query/tests/execute.rs

@@ -13,7 +13,7 @@
 #![cfg(feature = "sqlite")]
 
 use sqlx::{Row, SqlitePool};
-use sqlx_query::{Cursor, OrderByClause, QueryComposer, Value, WhereClause};
+use sqlx_query::{Cursor, OrderByClause, Pager, QueryComposer, Value, WhereClause};
 
 const SCHEMA: &str = "
     CREATE TABLE orders (
@@ -332,3 +332,57 @@ async fn a_cursor_page_stays_inside_the_base_condition() {
         "the page after acme's rank 20"
     );
 }
+
+/// Page through acme's orders `size` at a time by rank, carrying each
+/// page's cursor to the next as a token -- the way it travels between
+/// requests -- and return the ids of every page.
+async fn pages(pool: &SqlitePool, size: usize) -> Vec<Vec<i64>> {
+    let order_by = OrderByClause::default().asc("rank").asc("id");
+    let pager = Pager::new(Cursor::new(order_by.clone()), size);
+    let (mut seen, mut token) = (Vec::new(), None::<String>);
+
+    // Bounded, so a cursor that never runs out fails instead of hanging.
+    for _ in 0..10 {
+        let mut query = QueryComposer::<sqlx::Sqlite>::new(LIST_ORDERS);
+        query
+            .bind_value("acme")
+            .bind_value(pager.limit())
+            .push_order_by(order_by.clone());
+        if let Some(token) = &token {
+            query.with_cursor(Cursor::parse(token).expect("the token parses"));
+        }
+
+        let rows = query
+            .build()
+            .expect("composes")
+            .fetch_all(pool)
+            .await
+            .expect("runs");
+        let page = pager.next_page(rows).expect("the row carries every key");
+
+        seen.push(page.rows.iter().map(|row| row.get("id")).collect());
+        match page.cursor {
+            Some(cursor) => token = Some(cursor.encode()),
+            None => return seen,
+        }
+    }
+    panic!("still paging after ten pages: {seen:?}");
+}
+
+/// Every row once, in order, and the last page is the one with no cursor.
+#[tokio::test]
+async fn a_pager_pages_through_every_row_once() {
+    let pool = seed().await;
+
+    assert_eq!(pages(&pool, 2).await, [vec![1, 2], vec![3]]);
+    assert_eq!(pages(&pool, 1).await, [vec![1], vec![2], vec![3]]);
+}
+
+/// A page that exactly fills the size is the last: the extra row the pager
+/// asks for is what says there is another, and it did not come back.
+#[tokio::test]
+async fn a_page_that_fills_exactly_has_no_cursor() {
+    let pool = seed().await;
+
+    assert_eq!(pages(&pool, 3).await, [vec![1, 2, 3]]);
+}

crates/sqlx-query/tests/execute_postgres.rs

@@ -12,7 +12,7 @@
 #![cfg(feature = "postgres")]
 
 use sqlx::{PgPool, Row};
-use sqlx_query::{Cursor, OrderByClause, QueryComposer, WhereClause};
+use sqlx_query::{Cursor, OrderByClause, Pager, QueryComposer, WhereClause};
 
 /// `None` when there is no server to talk to, which is not the same as
 /// the variable being unset: the Nix dev shell exports the Dev Container's
@@ -319,3 +319,59 @@ async fn a_cursor_pages_on_a_uuid_key() {
     let rest: Vec<uuid::Uuid> = rows.iter().map(|row| row.get("id")).collect();
     assert_eq!(rest, ids[1..]);
 }
+
+/// The `$N` counterpart of SQLite's pager test: the pager's limit binds by
+/// number behind the cursor's values, and each page picks up after the
+/// last, across a tie the primary key breaks.
+#[tokio::test]
+async fn a_pager_pages_through_every_row_once() {
+    let pool = pool_or_skip!();
+    table(
+        &pool,
+        "pg_pager_test",
+        "id bigint primary key, rank bigint not null",
+    )
+    .await;
+
+    for (id, rank) in [(1i64, 30i64), (2, 10), (3, 20), (4, 20), (5, 40)] {
+        sqlx::query("INSERT INTO pg_pager_test VALUES ($1, $2)")
+            .bind(id)
+            .bind(rank)
+            .execute(&pool)
+            .await
+            .expect("insert");
+    }
+
+    let sql = "SELECT id, rank FROM pg_pager_test \
+               WHERE /* query.where AND */ TRUE ORDER BY /* query.order_by , */ id \
+               LIMIT $1";
+    let order_by = OrderByClause::default().asc("rank").asc("id");
+    let pager = Pager::new(Cursor::new(order_by.clone()), 2);
+    let (mut seen, mut cursor) = (Vec::<Vec<i64>>::new(), None);
+
+    for _ in 0..10 {
+        let mut query = QueryComposer::<sqlx::Postgres>::new(sql);
+        query
+            .bind_value(pager.limit())
+            .push_order_by(order_by.clone());
+        if let Some(cursor) = cursor.take() {
+            query.with_cursor(cursor);
+        }
+
+        let rows = query
+            .build()
+            .expect("composes")
+            .fetch_all(&pool)
+            .await
+            .expect("runs");
+        let page = pager.next_page(rows).expect("the row carries every key");
+
+        seen.push(page.rows.iter().map(|row| row.get("id")).collect());
+        match page.cursor {
+            Some(next) => cursor = Some(next),
+            None => break,
+        }
+    }
+
+    assert_eq!(seen, [vec![2, 3], vec![4, 1], vec![5]]);
+}