diff --git a/.changeset/keyset-pagination.md b/.changeset/keyset-pagination.md new file mode 100644 index 0000000..90379d8 --- /dev/null +++ b/.changeset/keyset-pagination.md @@ -0,0 +1,5 @@ +--- +"@maple-dev/effect-orm": minor +--- + +`/database` adds `paginate(query, { orderBy, after, limit }, params?)`: keyset (cursor) pagination over a built query. It adds a seek on the sort key, the `ORDER BY` and `LIMIT limit + 1`, and returns `{ rows, nextCursor }`, with no `OFFSET`. With one direction the seek is a row comparison, `(a, b) < (?, ?)`; with mixed directions it is the expanded `OR` form, bounded on the first key. The key is one or more selected, non-null aliases; end it with a unique column. The cursor is base64url text, encoded through the key columns' schemas. A cursor that does not decode, or that was written for another sort key, fails with the new `InvalidCursor` error. Postgres, ClickHouse, SQLite and D1. A nullable key is refused at run time too, since dialects sort NULLs differently. diff --git a/docs/database.md b/docs/database.md index bf14d00..fc7bbf8 100644 --- a/docs/database.md +++ b/docs/database.md @@ -84,6 +84,7 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti | `layerSqlClient(options)` | A `Database` layer over the `SqlClient` in context | | `Database` | The service | | `run(query, params?)` | Compile a query for the database's dialect, run it, decode its rows | +| `paginate(query, options, params?)` | One page of a query, by keyset, with a cursor for the next page ([Pagination](#pagination)) | | `sql\`...\`` | A statement with every `${value}` bound; `sql.identifier(name)` for a table or column name | | `query(statement, schema?)` | Run a statement and return its rows, decoded when a schema is given | | `execute(statement)` | Run a statement and discard its rows | @@ -155,6 +156,55 @@ Build the `Database` wherever you build the client. If each request has its own `Database` per request from it: nothing request-scoped is stored in the `Database`, and the open transaction lives in fiber context. +### Pagination + +`paginate` reads a query one page at a time by keyset. It adds a seek on the sort key, the +`ORDER BY` and `LIMIT limit + 1`, then returns `limit` rows and a cursor for the next page. The +extra row only tells if there is a next page. There is no `OFFSET`: the database goes to the +cursor through an index on the key, and does not read and drop all earlier rows. A row added +before the cursor does not move later pages. + +```ts +const page = yield* Db.paginate( + PG.from(Posts) + .select("id", "createdAt", "title") + .where(($) => [$.orgId.eq(PG.param.string("orgId"))]), + { orderBy: [["createdAt", "desc"], ["id", "desc"]], after: request.cursor, limit: 50 }, + { orgId }, +) // { rows: ReadonlyArray<{ id; createdAt; title }>, nextCursor: Option } +``` + +| Option | Meaning | +| --- | --- | +| `orderBy` | One or more selected aliases, each `"asc"` or `"desc"`. A nullable alias is a type error | +| `after` | The `nextCursor` of the previous page; leave it out for the first page | +| `limit` | Rows per page, a positive integer. Anything else fails with `QueryBuilderError` | + +The key must be unique, so end it with a unique column such as the primary key. `paginate` +cannot check this: rows that tie on the whole key can be skipped at a page boundary. + +With one direction the seek is a row comparison, `("createdAt", "id") < ($2, $3)`, which +Postgres and SQLite (D1 too) serve from an index on `(createdAt, id)` and ClickHouse compares +as a tuple. With mixed directions it is `a >= x AND (a > x OR (a = x AND b < y))`. + +A NULL key has no place in the seek, and dialects disagree on where NULLs sort: ascending, +SQLite puts them first, Postgres and ClickHouse last. So a nullable key is a type error, +and a defect at run time when the types were bypassed. + +The cursor is base64url text. It holds the sort key and the last row's key values, encoded +through the key columns' schemas, so a `DateTime.Utc` key comes back as the same instant. It +is not encrypted or signed: do not put a value in the key that the reader must not see. A +cursor that does not decode, or that was written for another `orderBy`, fails with +`InvalidCursor`. It usually comes from a request, so answer it as a bad request. + +The query keeps its own `where`, so the tenant filter and params still apply. `paginate` +replaces the query's `orderBy` and `limit`. A query with `offset`, `groupBy`, `having` or +`distinctOn`, or a key that is not a selected alias, is a defect. + +`paginate` uses the `Database` in context; it is not a method on `DatabaseApi`. Its types are +`PaginateOptions`, `Page`, `KeysetOrder`, `KeysetColumn` (an alias that can be a key) and +`KeysetDirection`. + ## Transactions ```ts diff --git a/docs/recipes.md b/docs/recipes.md index 3b892c9..0f2d42f 100644 --- a/docs/recipes.md +++ b/docs/recipes.md @@ -143,7 +143,8 @@ console.log(compiled.sql) // ... LIMIT 25 OFFSET 25 Validate page sizes and indices as bounded non-negative integers before constructing a query. An offset is not a cursor: concurrent inserts can shift pages, and large offsets still require work. For a raw event feed, use a fixed time range and a cursor over a unique, stable sort key -(such as timestamp plus event ID). The builder does not create or validate cursors for you. +(such as timestamp plus event ID). `Db.paginate` does this for a query that is not grouped; see +[Pagination](./database.md#pagination). ## Preserve large integer IDs diff --git a/docs/reference.md b/docs/reference.md index fe8f4eb..10985a4 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -512,8 +512,8 @@ placeholders, double-quoted identifiers. | `ClickHouseStatementFromString`, `splitTerminalClauses`, `maskLiteralsAndComments` | `/sql` | | Schema tooling: `isSchemaObject`, `makeSnapshot`, `renderSchema`, `diffSchemas`, `diffPgSchemas`, `fromDrizzleSnapshot`, snapshot and entity schemas (`PgEnumEntity`, `PgUniqueEntity`, …), `Hint`; for SQLite, `sqliteEntitiesOf`, `renderSqliteSchema`, `renderSqliteCreateTable`, `renderSqliteCreateIndex`, `diffSqliteSchemas`, `fromDrizzleSqliteSnapshot`, `SqliteMigrationOp`, `renderSqliteOp`, `renderSqliteMigrationSql`, `labelOfSqlite`, `rebuildName`, the `SqliteSnapshot` schema and the `SqliteTableEntity`, `SqliteColumnEntity`, `SqliteIndexEntity`, `SqliteUniqueEntity` schemas | `/schema` | | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors | `/migrate` | -| `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` | -| `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | +| `Database`, `run`, `paginate`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` | +| `DatabaseError`, `InvalidCursor`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | | `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` | | `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` | | `defineConfig`, `generate`, `check`, `pull`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit` | diff --git a/src/ch/compile.ts b/src/ch/compile.ts index cf4ca5b..23823c5 100644 --- a/src/ch/compile.ts +++ b/src/ch/compile.ts @@ -1289,7 +1289,7 @@ export function selectedAliasesOf(query: CHQuery): ReadonlyArray< } /** Evaluate a query's SELECT callback without compiling it. */ -function selectExprsOf(query: CHQuery): Record | undefined { +export function selectExprsOf(query: CHQuery): Record | undefined { return selectRecordOf(query._state, makeAccessor(query._state)) } @@ -1297,7 +1297,7 @@ function selectExprsOf(query: CHQuery): Record | * A query's SELECT list against accessor `$`: its callback's record, or for a * bare `select()` over a subquery or union, every alias that source selects. */ -function selectRecordOf(state: CHQueryState, $: any): Record | undefined { +export function selectRecordOf(state: CHQueryState, $: any): Record | undefined { if (state.selectAll) { const source = state.fromUnion ? unionExprsOf(state.fromUnion._state.queries) : state.fromQuery ? selectExprsOf(state.fromQuery) : undefined return source === undefined ? undefined : Object.fromEntries(Object.keys(source).map((alias) => [alias, $[alias]])) diff --git a/src/ch/expr.ts b/src/ch/expr.ts index 6c2f84e..8656756 100644 --- a/src/ch/expr.ts +++ b/src/ch/expr.ts @@ -584,9 +584,18 @@ export function makeExpr( arith>(fragment, "-", n, schema), mod: (n: R | Expr) => arith>(fragment, "%", n, schema), } + operands.set(self, operand) return self } +// How each expression writes a plain value on the right of a comparison, for +// SQL that compares outside the comparison methods: a row value `(a, b) > (x, y)`. +const operands = new WeakMap, (value: unknown) => SqlFragment>() + +/** A plain value as `expr`'s comparisons write it: through the column's codec when `expr` is a column. */ +export const operandOf = (expr: Expr, value: unknown): SqlFragment => + operands.get(expr)?.(value) ?? toFragment(value) + /** * An expression with no declared result type — {@link untypedExpr}'s * counterpart for a caller assembling its own fragment. diff --git a/src/database.ts b/src/database.ts index dc56aab..a502dd4 100644 --- a/src/database.ts +++ b/src/database.ts @@ -31,6 +31,7 @@ export { } from "./database/database" export { DatabaseError, + InvalidCursor, TenantScopeViolation, TransactionClosed, TransactionCommitFailed, @@ -39,6 +40,14 @@ export { TransactionUnsupported, type TransactionError, } from "./database/errors" +export { + paginate, + type KeysetColumn, + type KeysetDirection, + type KeysetOrder, + type Page, + type PaginateOptions, +} from "./database/paginate" export { QuerySummary, withQuerySummary } from "./database/tracing" export { sql, type SqlIdentifier, type SqlRawText, type SqlTemplate } from "./database/sql" export type { DialectTransactions, IsolationLevel, TransactionSettings } from "./ch/dialect" diff --git a/src/database/errors.ts b/src/database/errors.ts index 786ab5a..4ba45dd 100644 --- a/src/database/errors.ts +++ b/src/database/errors.ts @@ -82,3 +82,13 @@ export type TransactionError = | TransactionCommitFailed | TransactionRollbackFailed | TransactionClosed + +/** + * `paginate` was given a cursor it cannot read: not one it wrote, one written + * for another sort order, or one changed on the way. The cursor usually comes + * from a request, so this is an error to answer (a 400), not a defect. + */ +export class InvalidCursor extends Schema.TaggedError()("@maple-dev/effect-orm/InvalidCursor", { + message: Schema.String, + cause: Schema.Defect(), +}) {} diff --git a/src/database/paginate.sqlite.test.ts b/src/database/paginate.sqlite.test.ts new file mode 100644 index 0000000..ee09200 --- /dev/null +++ b/src/database/paginate.sqlite.test.ts @@ -0,0 +1,82 @@ +// `paginate` on SQLite and a local D1: the row comparison and the mixed-direction +// seek both walk every row once, timestamps (stored as text) round-trip through +// the cursor, and a nullable key is refused even when the type check is bypassed. + +import * as D1Client from "@effect/sql-d1/D1Client" +import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient" +import { Cause, DateTime, Effect, Exit, Layer, Option } from "effect" +import * as SqlClient from "effect/sql/SqlClient" +import { Miniflare } from "miniflare" +import { expect, it } from "vitest" +import * as Db from "../database" +import * as SQLite from "../sqlite" +import type { Dialect } from "../ch/dialect" + +const Posts = SQLite.table( + "posts", + { id: SQLite.integer, org: SQLite.text, at: SQLite.timestamp, note: SQLite.nullable(SQLite.text) }, + { tenantColumn: "org" }, +) +const byOrg = SQLite.from(Posts) + .select("id", "at", "note") + .where(($) => [$.org.eq(SQLite.param.string("org"))]) +type Row = { readonly id: number; readonly at: DateTime.Utc; readonly note: string | null } + +const walk = (orderBy: Db.PaginateOptions["orderBy"]) => + Effect.gen(function* () { + const pages: Array> = [] + let after: string | undefined + for (;;) { + const page = yield* Db.paginate(byOrg, { orderBy, limit: 4, after }, { org: "a" }) + pages.push(page.rows.map((row) => row.id)) + if (Option.isNone(page.nextCursor)) return pages + after = page.nextCursor.value + } + }) + +const pagesOn = (dialect: Dialect, client: Layer.Layer) => + Effect.runPromise( + Effect.gen(function* () { + const db = Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect, requireTenantScope: true }) + yield* db.execute({ sql: "CREATE TABLE posts (id INTEGER PRIMARY KEY, org TEXT NOT NULL, at TEXT NOT NULL, note TEXT)" }) + // Ten posts in org "a" whose timestamps tie in pairs, with NULL notes on + // the odd ids, and one post in org "b". + const rows = Array.from({ length: 10 }, (_, i) => ({ + id: i + 1, + org: "a", + at: DateTime.makeUnsafe(Date.UTC(2026, 0, 1, 0, Math.ceil((i + 1) / 2))), + note: (i + 1) % 2 === 0 ? `n${i + 1}` : null, + })) + yield* db.run(SQLite.insertInto(Posts).values([...rows, { id: 11, org: "b", at: rows[0]!.at, note: null }]).crossTenant()) + const run = (effect: Effect.Effect) => Effect.provideService(effect, Db.Database, db) + const desc = yield* run(walk([["at", "desc"], ["id", "desc"]])) + const mixed = yield* run(walk([["at", "asc"], ["id", "desc"]])) + const nullable = yield* run(Effect.exit(Db.paginate(byOrg, { orderBy: [["note", "asc"]] as any, limit: 4 }, { org: "a" }))) + return { desc, mixed, nullable } + }).pipe(Effect.provide(client)), + ) + +const check = ({ desc, mixed, nullable }: Awaited>) => { + expect(desc).toEqual([[10, 9, 8, 7], [6, 5, 4, 3], [2, 1]]) + expect(mixed).toEqual([[2, 1, 4, 3], [6, 5, 8, 7], [10, 9]]) + expect(Exit.isFailure(nullable) && Cause.hasDies(nullable.cause)).toBe(true) + expect(String(Exit.isFailure(nullable) ? Cause.squash(nullable.cause) : "")).toMatch(/can be NULL/) +} + +it("walks keyset pages on SQLite", async () => { + check(await pagesOn(SQLite.sqliteDialect, SqliteClient.layer({ filename: ":memory:" }))) +}) + +it("walks keyset pages on D1", async () => { + const mf = new Miniflare({ + modules: true, + script: "export default { fetch() { return new Response('paginate test') } }", + compatibilityDate: "2026-07-01", + d1Databases: ["DB"], + }) + try { + check(await pagesOn(SQLite.d1Dialect, D1Client.layer({ db: await mf.getD1Database("DB") }))) + } finally { + await mf.dispose() + } +}, 30_000) diff --git a/src/database/paginate.test-d.ts b/src/database/paginate.test-d.ts new file mode 100644 index 0000000..79b3b1a --- /dev/null +++ b/src/database/paginate.test-d.ts @@ -0,0 +1,35 @@ +// Type-level tests for `paginate`: the page's rows, and which keys it accepts. + +import { Effect } from "effect" +import { expectTypeOf } from "expect-type" +import * as PG from "../postgres" +import * as Db from "../database" + +const Items = PG.table("items", { + columns: { id: PG.int4, org: PG.text, note: PG.nullable(PG.text) }, + primaryKey: ["id"], +}) +const items = PG.from(Items) + .select("id", "note") + .where(($) => [$.org.eq(PG.param.string("org"))]) +type Row = { readonly id: number; readonly note: string | null } + +// The page holds the query's rows and an optional next cursor. +expectTypeOf(Db.paginate(items, { orderBy: [["id", "asc"]], limit: 10 }, { org: "a" })).toEqualTypeOf< + Effect.Effect< + Db.Page, + Db.InvalidCursor | Db.DatabaseError | PG.QueryBuilderError | PG.CompiledQueryDecodeError, + Db.Database + > +>() + +// Only a selected, non-null alias can be a key. +expectTypeOf>().toEqualTypeOf<"id">() +// @ts-expect-error -- a nullable column cannot order a keyset +Db.paginate(items, { orderBy: [["note", "asc"]], limit: 10 }, { org: "a" }) +// @ts-expect-error -- "org" is not selected +Db.paginate(items, { orderBy: [["org", "asc"]], limit: 10 }, { org: "a" }) +// @ts-expect-error -- the key needs at least one column +Db.paginate(items, { orderBy: [], limit: 10 }, { org: "a" }) +// @ts-expect-error -- a missing param is a type error, as for `run` +Db.paginate(items, { orderBy: [["id", "asc"]], limit: 10 }) diff --git a/src/database/paginate.test.ts b/src/database/paginate.test.ts new file mode 100644 index 0000000..c81cb78 --- /dev/null +++ b/src/database/paginate.test.ts @@ -0,0 +1,172 @@ +// `paginate` against PGlite: pages walked with the cursor each one returns +// cover every row once, in order, and the SQL is a keyset seek, not an OFFSET. + +import { PgliteClient } from "@effect/sql-pglite" +import { expect, layer } from "@effect/vitest" +import { Cause, DateTime, Effect, Exit, Layer, Option } from "effect" +import * as SqlClient from "effect/sql/SqlClient" +import * as PG from "../postgres" +import { postgresDialect } from "../pg/dialect" +import * as Db from "../database" + +const statements: Array = [] + +const Live = Layer.effect( + Db.Database, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + return Db.fromSqlClient(sql, { + dialect: postgresDialect, + observe: (statement) => Effect.sync(() => void statements.push(statement.sql)), + }) + }), +).pipe(Layer.provideMerge(PgliteClient.layer({ postgresqlconf: "timezone = 'UTC'" }))) + +const Posts = PG.table("posts", { + columns: { id: PG.int4, org: PG.text, at: PG.timestamptz, title: PG.text, note: PG.nullable(PG.text) }, + primaryKey: ["id"], + tenantColumn: "org", +}) + +// Ten posts in org "a", with timestamps that tie in pairs, and one in org "b". +const seed = Effect.gen(function* () { + yield* Db.execute(Db.sql`CREATE TABLE IF NOT EXISTS posts (id int4 PRIMARY KEY, org text NOT NULL, at timestamptz NOT NULL, title text NOT NULL, note text)`) + for (let id = 1; id <= 10; id++) { + const at = new Date(Date.UTC(2026, 0, 1, 0, Math.ceil(id / 2))) + yield* Db.execute(Db.sql`INSERT INTO posts VALUES (${id}, ${"a"}, ${at.toISOString()}, ${`t${id}`}, NULL) ON CONFLICT DO NOTHING`) + } + yield* Db.execute(Db.sql`INSERT INTO posts VALUES (${11}, ${"b"}, ${"2026-01-01T00:03:00Z"}, ${"other"}, NULL) ON CONFLICT DO NOTHING`) +}) + +const byOrg = PG.from(Posts) + .select("id", "at", "title") + .where(($) => [$.org.eq(PG.param.string("org"))]) + +/** Every page, following `nextCursor` until there is none. */ +const walk = >( + options: Omit, +) => + Effect.gen(function* () { + const pages: Array> = [] + let after: string | undefined + for (;;) { + const page = yield* Db.paginate(byOrg, { ...options, after } as O, { org: "a" }) + pages.push(page.rows.map((row) => row.id)) + if (Option.isNone(page.nextCursor)) return pages + after = page.nextCursor.value + } + }) + +layer(Live, { excludeTestServices: true })("paginate on PGlite", (it) => { + it.effect("pages walked by cursor cover every row once, newest first, ties broken by id", () => + Effect.gen(function* () { + yield* seed + const pages = yield* walk({ orderBy: [["at", "desc"], ["id", "desc"]], limit: 4 }) + expect(pages).toEqual([[10, 9, 8, 7], [6, 5, 4, 3], [2, 1]]) + }), + ) + + it.effect("a page is a seek: a row comparison, ORDER BY the key and LIMIT n + 1, with no OFFSET", () => + Effect.gen(function* () { + yield* seed + const first = yield* Db.paginate(byOrg, { orderBy: [["at", "desc"], ["id", "desc"]], limit: 4 }, { org: "a" }) + statements.length = 0 + yield* Db.paginate(byOrg, { orderBy: [["at", "desc"], ["id", "desc"]], limit: 4, after: Option.getOrThrow(first.nextCursor) }, { org: "a" }) + const sent = statements[0]! + expect(sent).toContain(`("posts"."at", "posts"."id") < ($2, $3)`) + expect(sent).toMatch(/ORDER BY "posts"\."at" DESC, "posts"\."id" DESC\s+LIMIT 5$/) + expect(sent).not.toContain("OFFSET") + }), + ) + + it.effect("mixed directions seek with the expanded OR form, bounded on the first key", () => + Effect.gen(function* () { + yield* seed + const pages = yield* walk({ orderBy: [["at", "asc"], ["id", "desc"]], limit: 3 }) + expect(pages).toEqual([[2, 1, 4], [3, 6, 5], [8, 7, 10], [9]]) + expect(statements.at(-1)).toContain( + `"posts"."at" >= $2 AND ("posts"."at" > $2 OR ("posts"."at" = $2 AND "posts"."id" < $3))`, + ) + }), + ) + + it.effect("one key compares on its own, and a page that ends the rows has no next cursor", () => + Effect.gen(function* () { + yield* seed + expect(yield* walk({ orderBy: [["id", "asc"]], limit: 5 })).toEqual([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]) + expect(statements.at(-1)).toContain(`"posts"."id" > $2`) + }), + ) + + it.effect("the query keeps its own where: the tenant filter and params still apply", () => + Effect.gen(function* () { + yield* seed + const page = yield* Db.paginate(byOrg, { orderBy: [["id", "asc"]], limit: 20 }, { org: "b" }) + expect(page.rows.map((row) => row.title)).toEqual(["other"]) + expect(page.nextCursor).toEqual(Option.none()) + }), + ) + + it.effect("a key value round-trips through its codec: a cursor on a timestamp seeks to the same instant", () => + Effect.gen(function* () { + yield* seed + const first = yield* Db.paginate(byOrg, { orderBy: [["at", "asc"], ["id", "asc"]], limit: 1 }, { org: "a" }) + expect(DateTime.formatIso(first.rows[0]!.at)).toBe("2026-01-01T00:01:00.000Z") + const next = yield* Db.paginate( + byOrg, + { orderBy: [["at", "asc"], ["id", "asc"]], limit: 1, after: Option.getOrThrow(first.nextCursor) }, + { org: "a" }, + ) + expect(next.rows.map((row) => row.id)).toEqual([2]) + }), + ) + + it.effect("a cursor that is not one paginate wrote fails with InvalidCursor", () => + Effect.gen(function* () { + yield* seed + const orderBy = [["id", "asc"]] as const + for (const after of ["not a cursor", "eyJrIjoiaWQgYXNjIiwidiI6WyJ4Il19", ""]) { + const error = yield* Effect.flip(Db.paginate(byOrg, { orderBy, limit: 2, after }, { org: "a" })) + expect(error).toBeInstanceOf(Db.InvalidCursor) + } + }), + ) + + it.effect("a cursor written for another sort key fails with InvalidCursor", () => + Effect.gen(function* () { + yield* seed + const first = yield* Db.paginate(byOrg, { orderBy: [["id", "asc"]], limit: 2 }, { org: "a" }) + const error = yield* Effect.flip( + Db.paginate(byOrg, { orderBy: [["id", "desc"]], limit: 2, after: Option.getOrThrow(first.nextCursor) }, { org: "a" }), + ) + expect(error._tag).toBe("@maple-dev/effect-orm/InvalidCursor") + }), + ) + + it.effect("a limit that is not a positive integer is a QueryBuilderError", () => + Effect.gen(function* () { + for (const limit of [0, -1, 2.5, Number.NaN]) { + const error = yield* Effect.flip(Db.paginate(byOrg, { orderBy: [["id", "asc"]], limit }, { org: "a" })) + expect(error).toBeInstanceOf(PG.QueryBuilderError) + } + }), + ) + + it.effect("a query paginate cannot seek is a defect: offset, groupBy, a key not selected", () => + Effect.gen(function* () { + const defects: Array> = [ + Db.paginate(byOrg.offset(10), { orderBy: [["id", "asc"]], limit: 2 }, { org: "a" }), + Db.paginate( + PG.from(Posts).select(($) => ({ org: $.org, n: PG.count() })).groupBy("org"), + { orderBy: [["org", "asc"]], limit: 2 }, + ), + // @ts-expect-error -- "org" is not selected + Db.paginate(byOrg, { orderBy: [["org", "asc"]], limit: 2 }, { org: "a" }), + ] + for (const effect of defects) { + const exit = yield* Effect.exit(effect) + expect(Exit.isFailure(exit) && Cause.hasDies(exit.cause)).toBe(true) + } + }), + ) +}) diff --git a/src/database/paginate.ts b/src/database/paginate.ts new file mode 100644 index 0000000..fc6e30c --- /dev/null +++ b/src/database/paginate.ts @@ -0,0 +1,187 @@ +// Keyset (cursor) pagination over a built query. +// +// A page is the query with a seek added: `WHERE (a, b) > (?, ?)` from the +// cursor, `ORDER BY a, b` and `LIMIT n + 1`. The extra row says whether there is +// a next page and is not returned. The cursor holds the last row's key values, +// encoded through the key columns' own schemas, so reading one back is a +// decode: a cursor the caller changed is an `InvalidCursor`, not bad SQL. +// +// Unlike OFFSET, the database seeks to the cursor through an index on the key +// instead of reading and dropping every earlier row, and a row inserted ahead +// of the cursor does not shift later pages. See docs/database.md#pagination. + +import { Effect, Option, Schema } from "effect" +import { selectExprsOf, selectRecordOf, type CompiledQueryDecodeError } from "../ch/compile" +import { QueryBuilderDefect, QueryBuilderError } from "../ch/errors" +import { and, isExprLike, makeCond, operandOf, or, type Condition, type Expr, type ParamsSatisfied } from "../ch/expr" +import type { CHQuery } from "../ch/query" +import type { QueryParams } from "../ch/union" +import { compile, known } from "../sql/sql-fragment" +import { Database, type RowOf, type RunCheck } from "./database" +import { InvalidCursor, type DatabaseError } from "./errors" + +export type KeysetDirection = "asc" | "desc" + +/** + * An output alias that can be part of a keyset: one that is never NULL. A NULL + * key has no place in `(a, b) > (?, ?)`, so the rows holding one would be + * skipped or repeated. + */ +export type KeysetColumn = { + [K in keyof Row & string]-?: null extends Row[K] ? never : undefined extends Row[K] ? never : K +}[keyof Row & string] + +/** The sort key of a page: one or more selected aliases, each with its direction. */ +export type KeysetOrder = readonly [ + readonly [KeysetColumn, KeysetDirection], + ...ReadonlyArray, KeysetDirection]>, +] + +export interface PaginateOptions { + /** + * The sort key. Together its columns must be unique, so end it with a unique + * column (the primary key) as a tiebreaker: rows that tie on the whole key + * can be skipped at a page boundary. `paginate` cannot check this. + */ + readonly orderBy: KeysetOrder + /** The `nextCursor` of the previous page; leave out for the first page. */ + readonly after?: string | undefined + /** Rows per page: a positive integer. */ + readonly limit: number +} + +export interface Page { + readonly rows: ReadonlyArray + /** The cursor for the next page, or `none` when this page is the last. */ + readonly nextCursor: Option.Option +} + +const defect = (message: string) => Effect.die(new QueryBuilderDefect({ message: `paginate: ${message}` })) + +/** + * The cursor: base64url of `{ k, v }`, where `k` names the sort key and `v` + * holds the key values in their JSON form. A cursor for another sort key fails + * on `k`, so it cannot seek on the wrong columns. + */ +const cursorCodec = (sortKey: string, schemas: ReadonlyArray) => + Schema.StringFromBase64Url.pipe( + Schema.decodeTo( + Schema.fromJsonString( + Schema.toCodecJson( + Schema.Struct({ + k: Schema.Literal(sortKey), + v: Schema.Tuple(schemas.map((schema) => Schema.toType(schema))), + }), + ), + ), + ), + ) as unknown as Schema.Codec<{ readonly k: string; readonly v: ReadonlyArray }, string> + +const after = (expr: Expr, direction: KeysetDirection, value: unknown): Condition => + direction === "asc" ? expr.gt(value) : expr.lt(value) + +/** + * The rows after `values` in the order of `exprs`. With one direction it is a + * row comparison, `(a, b) > (x, y)`, which both dialects compare in order and + * Postgres serves from an index on `(a, b)`. With mixed directions it is + * `a >= x AND (a > x OR (a = x AND b < y))`: the leading bound keeps the seek + * on the index's first column. + */ +const seek = ( + exprs: ReadonlyArray>, + directions: ReadonlyArray, + values: ReadonlyArray, +): Condition => { + if (exprs.length === 1) return after(exprs[0]!, directions[0]!, values[0]) + if (directions.every((direction) => direction === directions[0])) { + return makeCond( + known(() => { + const left = exprs.map((expr) => compile(expr.toFragment())).join(", ") + const right = exprs.map((expr, i) => compile(operandOf(expr, values[i]))).join(", ") + return `(${left}) ${directions[0] === "asc" ? ">" : "<"} (${right})` + }), + ) + } + const first = directions[0] === "asc" ? exprs[0]!.gte(values[0]) : exprs[0]!.lte(values[0]) + const branches = exprs.map( + (expr, i) => and(...exprs.slice(0, i).map((prefix, j) => prefix.eq(values[j])), after(expr, directions[i]!, values[i]))!, + ) + return and(first, or(...branches)) +} + +/** + * One page of `query`, in `orderBy` order, after the `after` cursor. + * + * The query keeps its own `where`; `paginate` adds the seek and replaces its + * `orderBy` and `limit`. A query with `offset`, `groupBy`, `having` or + * `distinctOn` is refused: a keyset seeks over rows, not over groups. + */ +export const paginate = , const Given extends object = {}>( + query: Q & RunCheck & ParamsSatisfied, Given>, + options: PaginateOptions>, + params?: Given, +): Effect.Effect< + Page>, + InvalidCursor | DatabaseError | QueryBuilderError | CompiledQueryDecodeError, + Database +> => + Effect.gen(function* () { + const state = query._state + const orderBy: ReadonlyArray = options.orderBy + if (orderBy.length === 0) return yield* defect("orderBy needs at least one key") + if (state.offsetValue !== undefined) return yield* defect("the cursor replaces offset(); remove it") + if (state.groupByKeys.length > 0 || state.havingFn !== undefined) { + return yield* defect("a keyset pages over rows; a query with groupBy or having is not supported") + } + if (Array.isArray(state.distinct)) return yield* defect("distinctOn sets its own ORDER BY; it cannot be paged by a keyset") + if (!Number.isInteger(options.limit) || options.limit < 1) { + return yield* new QueryBuilderError({ + code: "InvalidArguments", + message: `paginate: limit must be a positive integer, not ${String(options.limit)}`, + }) + } + + const selected = selectExprsOf(query) ?? {} + const keys = orderBy.map(([key]) => key) + if (new Set(keys).size !== keys.length) return yield* defect("orderBy names a key twice") + const schemas: Array = [] + for (const key of keys) { + const expr = selected[key] + if (!isExprLike(expr)) return yield* defect(`${JSON.stringify(key)} is not a selected alias`) + if (expr.schema === undefined) return yield* defect(`${JSON.stringify(key)} has no type, so its cursor value cannot be encoded`) + // The type check already refuses this; an untyped caller is caught here. + // A NULL key drops out of the seek, and dialects disagree on where NULLs + // sort (ascending, SQLite puts them first, Postgres and ClickHouse last). + if (Schema.is(Schema.toType(expr.schema))(null)) { + return yield* defect(`${JSON.stringify(key)} can be NULL, and a NULL key has no place in the seek`) + } + schemas.push(expr.schema) + } + const sortKey = orderBy.map(([key, direction]) => `${key} ${direction}`).join(",") + const codec = cursorCodec(sortKey, schemas) + + const from = + options.after === undefined + ? undefined + : yield* Schema.decodeUnknownEffect(codec)(options.after).pipe( + Effect.mapError((cause) => new InvalidCursor({ message: "paginate: the cursor is not one this query wrote", cause })), + ) + + const keyExprs = ($: any): ReadonlyArray> => { + const record = selectRecordOf(state, $) ?? {} + return keys.map((key) => record[key]) + } + const directions = orderBy.map(([, direction]) => direction) + const paged = query + .where(($) => [from === undefined ? undefined : seek(keyExprs($), directions, from.v)]) + .orderBy(($) => keyExprs($).map((expr, i) => [expr, directions[i]!] as const)) + .limit(options.limit + 1) + + const db = yield* Database + const rows = (yield* db.run(paged as any, params)) as ReadonlyArray> + if (rows.length <= options.limit) return { rows, nextCursor: Option.none() } + const page = rows.slice(0, options.limit) + const last = page[page.length - 1] as Record + const cursor = yield* Schema.encodeEffect(codec)({ k: sortKey, v: keys.map((key) => last[key]) }).pipe(Effect.orDie) + return { rows: page, nextCursor: Option.some(cursor) } + }) diff --git a/tests/paginate.clickhouse.test.ts b/tests/paginate.clickhouse.test.ts new file mode 100644 index 0000000..823d2fd --- /dev/null +++ b/tests/paginate.clickhouse.test.ts @@ -0,0 +1,84 @@ +// `paginate` on a live ClickHouse: a DateTime64 + UInt64 key, in one direction +// (a tuple comparison) and in mixed directions (the expanded OR form). Creates +// one table in a throwaway database. + +import { ClickhouseClient } from "@effect/sql-clickhouse" +import { Effect, Option } from "effect" +import { describe, expect, it } from "vitest" +import * as CH from "@maple-dev/effect-orm/clickhouse" +import * as Db from "@maple-dev/effect-orm/database" +import { endpoint } from "./clickhouse-support" + +const user = process.env.EFFECT_ORM_CLICKHOUSE_USER ?? "default" +const password = process.env.EFFECT_ORM_CLICKHOUSE_PASSWORD ?? "" + +const Events = CH.table("events", { + external: true, + tenantColumn: "OrgId", + columns: { OrgId: CH.string, At: CH.dateTime64, Id: CH.uint64 }, +}) +const byOrg = CH.from(Events) + .select("Id", "At") + .where(($) => [$.OrgId.eq(CH.param.string("org"))]) + +const withEvents = (body: (db: Db.DatabaseApi, sent: Array) => Effect.Effect) => + Effect.gen(function* () { + const database = `eo_paginate_${Date.now()}_${Math.floor(Math.random() * 1e6)}` + const admin = yield* ClickhouseClient.ClickhouseClient + yield* admin.asCommand(admin.unsafe(`CREATE DATABASE ${database}`)) + return yield* Effect.gen(function* () { + const client = yield* ClickhouseClient.ClickhouseClient + const sent: Array = [] + const db = Db.fromSqlClient(client, { + dialect: CH.clickhouseDialect, + command: client.asCommand, + observe: (statement) => Effect.sync(() => void sent.push(statement.sql)), + }) + yield* db.execute(Db.sql`CREATE TABLE events (OrgId String, At DateTime64(3), Id UInt64) ENGINE = MergeTree ORDER BY (OrgId, At, Id)`) + // Ids 1..7; timestamps tie in pairs and carry milliseconds. + yield* db.run( + CH.insertInto(Events).values([ + ...[1, 2, 3, 4, 5, 6, 7].map((id) => ({ + OrgId: "a", + At: new Date(Date.UTC(2026, 0, 1, 0, 0, Math.ceil(id / 2), 250)), + Id: id, + })), + { OrgId: "b", At: new Date(Date.UTC(2026, 0, 1)), Id: 8 }, + ]), + ) + return yield* body(db, sent).pipe(Effect.provideService(Db.Database, db)) + }).pipe( + Effect.provide(ClickhouseClient.layer({ url: endpoint!, username: user, password, database })), + Effect.ensuring(Effect.orDie(admin.asCommand(admin.unsafe(`DROP DATABASE IF EXISTS ${database} SYNC`)))), + ) + }).pipe(Effect.provide(ClickhouseClient.layer({ url: endpoint!, username: user, password, database: "default" }))) + +const walk = (orderBy: Db.KeysetOrder>, limit: number) => + Effect.gen(function* () { + const pages: Array> = [] + let after: string | undefined + for (;;) { + const page = yield* Db.paginate(byOrg, { orderBy, limit, after }, { org: "a" }) + pages.push(page.rows.map((row) => row.Id)) + if (Option.isNone(page.nextCursor)) return pages + after = page.nextCursor.value + } + }) + +describe("paginate", () => { + describe.skipIf(!endpoint)("live ClickHouse", () => { + it("walks a DateTime64 + UInt64 key newest first with a tuple comparison", async () => { + const { pages, sent } = await Effect.runPromise( + withEvents((_db, sent) => Effect.map(walk([["At", "desc"], ["Id", "desc"]], 3), (pages) => ({ pages, sent }))), + ) + expect(pages).toEqual([[7, 6, 5], [4, 3, 2], [1]]) + expect(sent.at(-1)).toMatch(/\(events\.At, events\.Id\) < \('2026-01-01 00:00:01\.250', 2\)/) + expect(sent.at(-1)).toMatch(/ORDER BY events\.At DESC, events\.Id DESC\s+LIMIT 4/) + }) + + it("walks mixed directions with the expanded OR form", async () => { + const pages = await Effect.runPromise(withEvents(() => walk([["At", "asc"], ["Id", "desc"]], 2))) + expect(pages).toEqual([[2, 1], [4, 3], [6, 5], [7]]) + }) + }) +})