diff --git a/README.md b/README.md index cfe6bc3..6ae6e21 100644 --- a/README.md +++ b/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::::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. diff --git a/crates/sqlx-query/src/clauses/mod.rs b/crates/sqlx-query/src/clauses/mod.rs index aa1d1f5..459ccc9 100644 --- a/crates/sqlx-query/src/clauses/mod.rs +++ b/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; diff --git a/crates/sqlx-query/src/clauses/pager.rs b/crates/sqlx-query/src/clauses/pager.rs index 51f1323..d04488d 100644 --- a/crates/sqlx-query/src/clauses/pager.rs +++ b/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(&self, mut rows: Vec) -> Result, CursorError> + where + R: Row, + usize: ColumnIndex, + for<'r> bool: Decode<'r, R::Database> + Type, + for<'r> i16: Decode<'r, R::Database> + Type, + for<'r> i32: Decode<'r, R::Database> + Type, + for<'r> i64: Decode<'r, R::Database> + Type, + for<'r> f32: Decode<'r, R::Database> + Type, + for<'r> f64: Decode<'r, R::Database> + Type, + for<'r> String: Decode<'r, R::Database> + Type, + for<'r> crate::value::TimestampRepr: Decode<'r, R::Database> + Type, + for<'r> Vec: Decode<'r, R::Database> + Type, + for<'r> crate::value::UuidRepr: Decode<'r, R::Database> + Type, + { + 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 { + /// The page's rows, at most the pager's size. + pub rows: Vec, + /// The cursor to the next page — [`encode`](Cursor::encode) it as the + /// `next_page_token` — or `None` on the last page. + pub cursor: Option, +} + /// `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(); diff --git a/crates/sqlx-query/src/lib.rs b/crates/sqlx-query/src/lib.rs index b968d65..56dd4ee 100644 --- a/crates/sqlx-query/src/lib.rs +++ b/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}; diff --git a/crates/sqlx-query/tests/execute.rs b/crates/sqlx-query/tests/execute.rs index 4d5b87d..4d4ef09 100644 --- a/crates/sqlx-query/tests/execute.rs +++ b/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> { + 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::); + + // Bounded, so a cursor that never runs out fails instead of hanging. + for _ in 0..10 { + let mut query = QueryComposer::::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]]); +} diff --git a/crates/sqlx-query/tests/execute_postgres.rs b/crates/sqlx-query/tests/execute_postgres.rs index 0b7e217..ad8f6f1 100644 --- a/crates/sqlx-query/tests/execute_postgres.rs +++ b/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 = 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::>::new(), None); + + for _ in 0..10 { + let mut query = QueryComposer::::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]]); +}