From 3407badd7f9e792560372d26d5648576f01e233a Mon Sep 17 00:00:00 2001 From: Taras Kornichuk Date: Sun, 25 Oct 2026 17:17:04 +0200 Subject: [PATCH 1/3] feat(database): batch lookups by key with findById and findAllBy --- .changeset/batched-lookups.md | 5 + src/database.ts | 8 ++ src/database/database.ts | 3 +- src/database/resolver.test-d.ts | 39 +++++++ src/database/resolver.test.ts | 152 ++++++++++++++++++++++++++ src/database/resolver.ts | 187 ++++++++++++++++++++++++++++++++ 6 files changed, 393 insertions(+), 1 deletion(-) create mode 100644 .changeset/batched-lookups.md create mode 100644 src/database/resolver.test-d.ts create mode 100644 src/database/resolver.test.ts create mode 100644 src/database/resolver.ts diff --git a/.changeset/batched-lookups.md b/.changeset/batched-lookups.md new file mode 100644 index 0000000..44afe2a --- /dev/null +++ b/.changeset/batched-lookups.md @@ -0,0 +1,5 @@ +--- +"@maple-dev/effect-orm": minor +--- + +`Db.findById(query, { key })` and `Db.findAllBy(query, { key })` in `/database` batch lookups by a key with Effect's `RequestResolver`: N concurrent calls send one statement, the query with `key IN (...)` added to its WHERE, and each row decodes through the query's row schema. A batch never mixes `Database`s, transactions or params, so a tenant filter in the query (a value or a `param.*`, given with `.with(params)`) keeps each statement to one tenant, and `requireTenantScope` checks it. diff --git a/src/database.ts b/src/database.ts index ba66164..da148e3 100644 --- a/src/database.ts +++ b/src/database.ts @@ -39,5 +39,13 @@ export { type TransactionError, } from "./database/errors" export { QuerySummary, withQuerySummary } from "./database/tracing" +export { + findAllBy, + findById, + type FindAllBy, + type FindById, + type LookupError, + type LookupOptions, +} from "./database/resolver" export { sql, type SqlIdentifier, type SqlRawText, type SqlTemplate } from "./database/sql" export type { DialectTransactions, IsolationLevel, TransactionSettings } from "./ch/dialect" diff --git a/src/database/database.ts b/src/database/database.ts index aaf89d4..c912338 100644 --- a/src/database/database.ts +++ b/src/database/database.ts @@ -170,7 +170,8 @@ interface OpenState { open: boolean readonly client: SqlClient.SqlClient } -class TransactionOpen extends Context.Service()("@maple-dev/effect-orm/TransactionOpen") {} +/** @internal — also read by `resolver.ts` to keep batches inside one transaction. */ +export class TransactionOpen extends Context.Service()("@maple-dev/effect-orm/TransactionOpen") {} // A defect from the body, carried through `withTransaction` as a failure so the // transaction still rolls back and any defect that comes out of it is known to diff --git a/src/database/resolver.test-d.ts b/src/database/resolver.test-d.ts new file mode 100644 index 0000000..df020ce --- /dev/null +++ b/src/database/resolver.test-d.ts @@ -0,0 +1,39 @@ +// Type-level tests: what `findById` and `findAllBy` take and give. + +import type { Cause, Effect } from "effect" +import { expectTypeOf } from "expect-type" +import * as CH from "../ch/index" +import * as PG from "../postgres" +import * as Db from "../database" + +const Users = PG.table("users", { columns: { id: PG.int4, org: PG.text, name: PG.text }, primaryKey: ["id"] }) +type Row = { readonly id: number; readonly name: string } + +const userById = Db.findById(CH.from(Users).select("id", "name"), { key: "id" }) +expectTypeOf(userById(1)).toEqualTypeOf>() +// @ts-expect-error -- the key is the column's type +userById("1") + +const usersByName = Db.findAllBy(CH.from(Users).select("id", "name"), { key: "name" }) +expectTypeOf(usersByName("Ann")).toEqualTypeOf, Db.LookupError, Db.Database>>() + +// The key must be a selected column of the FROM table. +// @ts-expect-error -- `org` is not selected +Db.findById(CH.from(Users).select("id", "name"), { key: "org" }) +// @ts-expect-error -- not a column +Db.findById(CH.from(Users).select("id", "name"), { key: "nope" }) + +// A query with params takes them through `with`, checked as `run` checks them. +const inOrg = Db.findById( + CH.from(Users) + .select("id", "name") + .where(($) => [$.org.eq(CH.param.string("org"))]), + { key: "id" }, +) +expectTypeOf(inOrg.with({ org: "a" })(1)).toEqualTypeOf< + Effect.Effect +>() +// @ts-expect-error -- the params are required +inOrg(1) +// @ts-expect-error -- a param of the wrong type +inOrg.with({ org: 1 }) diff --git a/src/database/resolver.test.ts b/src/database/resolver.test.ts new file mode 100644 index 0000000..c93cc3a --- /dev/null +++ b/src/database/resolver.test.ts @@ -0,0 +1,152 @@ +// Request batching: `findById` and `findAllBy` turn N concurrent lookups into +// one `WHERE key IN (...)` statement. Counted on PGlite through `observe`. + +import { PgliteClient } from "@effect/sql-pglite" +import { assert, expect, layer } from "@effect/vitest" +import { Cause, Effect, Exit, Layer, Result } from "effect" +import * as SqlClient from "effect/sql/SqlClient" +import * as CH from "../ch/index" +import * as PG from "../postgres" +import { postgresDialect } from "../pg/dialect" +import * as Db from "../database" + +const statements: Array = [] + +const DatabaseLive = 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 Users = PG.table("users", { + columns: { id: PG.int4, org: PG.text, name: PG.text }, + primaryKey: ["id"], + tenantColumn: "org", +}) + +const Live = Layer.effectDiscard( + Effect.gen(function* () { + yield* Db.execute(Db.sql`CREATE TABLE users (id int4 PRIMARY KEY, org text NOT NULL, name text NOT NULL)`) + yield* Db.execute(Db.sql`INSERT INTO users VALUES (1, 'a', 'Ann'), (2, 'a', 'Bob'), (3, 'b', 'Cy')`) + }), +).pipe(Layer.provideMerge(DatabaseLive)) + +/** Run `effect` and return the statements it sent. */ +const counted = (effect: Effect.Effect) => + Effect.gen(function* () { + statements.length = 0 + const value = yield* effect + return { value, sent: [...statements] } + }) + +layer(Live, { excludeTestServices: true })("Request batching on PGlite", (it) => { + it.effect("findById sends one statement for N concurrent lookups and decodes each row", () => + Effect.gen(function* () { + const userById = Db.findById(CH.from(Users).select("id", "name").crossTenant(), { key: "id" }) + const { value, sent } = yield* counted(Effect.forEach([1, 2, 3], userById, { concurrency: "unbounded" })) + expect(value).toEqual([ + { id: 1, name: "Ann" }, + { id: 2, name: "Bob" }, + { id: 3, name: "Cy" }, + ]) + assert.strictEqual(sent.length, 1) + expect(sent[0]).toContain(" IN (") + }), + ) + + it.effect("findById fails a missing key with NoSuchElementError and sends a repeated key once", () => + Effect.gen(function* () { + const userById = Db.findById(CH.from(Users).select("id", "name").crossTenant(), { key: "id" }) + const { value, sent } = yield* counted( + Effect.forEach([1, 1, 9], (id) => Effect.result(userById(id)), { concurrency: "unbounded" }), + ) + assert(Result.isSuccess(value[0]) && Result.isSuccess(value[1])) + expect(value[1].success).toEqual({ id: 1, name: "Ann" }) + assert(Result.isFailure(value[2])) + expect(value[2].failure).toBeInstanceOf(Cause.NoSuchElementError) + assert.strictEqual(sent.length, 1) + expect(sent[0]).toMatch(/IN \(\$1, \$2\)/) + }), + ) + + it.effect("one batch never mixes tenants: each params value gets its own single-tenant statement", () => + Effect.gen(function* () { + const strict = Db.fromSqlClient(yield* SqlClient.SqlClient, { + dialect: postgresDialect, + requireTenantScope: true, + observe: (statement) => Effect.sync(() => void statements.push(statement.sql)), + }) + const inOrg = CH.from(Users) + .select("id", "name") + .where(($) => [$.org.eq(CH.param.string("org"))]) + const userById = Db.findById(inOrg, { key: "id" }) + const lookups = [ + userById.with({ org: "a" })(1), + userById.with({ org: "b" })(3), + userById.with({ org: "a" })(2), + // Another tenant's row is not found, even with its id. + Effect.flip(userById.with({ org: "b" })(1)), + ] as const + const { value, sent } = yield* counted( + Effect.all(lookups, { concurrency: "unbounded" }).pipe(Effect.provideService(Db.Database, strict)), + ) + expect(value.slice(0, 3)).toEqual([ + { id: 1, name: "Ann" }, + { id: 3, name: "Cy" }, + { id: 2, name: "Bob" }, + ]) + expect(value[3]).toBeInstanceOf(Cause.NoSuchElementError) + assert.strictEqual(sent.length, 2) + + // Without a tenant filter, requireTenantScope refuses the batch statement. + const unscoped = Db.findById(CH.from(Users).select("id", "name"), { key: "id" }) + const exit = yield* Effect.exit(unscoped(1).pipe(Effect.provideService(Db.Database, strict))) + assert(Exit.isFailure(exit)) + expect(Cause.squash(exit.cause)).toBeInstanceOf(Db.TenantScopeViolation) + }), + ) + + it.effect("a lookup inside a transaction never shares a batch with one outside it", () => + Effect.gen(function* () { + const userById = Db.findById(CH.from(Users).select("id", "name").crossTenant(), { key: "id" }) + const { value, sent } = yield* counted( + Effect.all([Db.transaction(userById(1)), userById(2)], { concurrency: "unbounded" }), + ) + expect(value).toEqual([ + { id: 1, name: "Ann" }, + { id: 2, name: "Bob" }, + ]) + assert.strictEqual(sent.length, 2) + }), + ) + + it.effect("findAllBy groups every row under its key, and a key with no rows gets an empty array", () => + Effect.gen(function* () { + const usersInOrg = Db.findAllBy(CH.from(Users).select("id", "org").orderBy(["id", "asc"]).crossTenant(), { key: "org" }) + const { value, sent } = yield* counted(Effect.forEach(["a", "b", "z"], usersInOrg, { concurrency: "unbounded" })) + expect(value).toEqual([ + [ + { id: 1, org: "a" }, + { id: 2, org: "a" }, + ], + [{ id: 3, org: "b" }], + [], + ]) + assert.strictEqual(sent.length, 1) + }), + ) + + it.effect("refuses a query with limit or offset, which would apply to the whole batch", () => + Effect.gen(function* () { + const firstUser = Db.findById(CH.from(Users).select("id", "name").limit(1).crossTenant(), { key: "id" }) + const { value, sent } = yield* counted(Effect.flip(firstUser(1))) + expect(value).toBeInstanceOf(CH.QueryBuilderError) + assert.strictEqual(sent.length, 0) + }), + ) +}) diff --git a/src/database/resolver.ts b/src/database/resolver.ts new file mode 100644 index 0000000..40592dd --- /dev/null +++ b/src/database/resolver.ts @@ -0,0 +1,187 @@ +// Request batching over Effect's `RequestResolver`: N concurrent lookups by a +// key become one `WHERE key IN (...)` statement, run with `Database.run`, so +// rows decode through the query's own row schema and `requireTenantScope` +// checks the statement that is sent. See docs/database.md#batching-lookups. +// +// Effect's `SqlResolver` keys batches by the `SqlClient` transaction found in +// context. A `Database` does not need a `SqlClient` in context, so the batch +// key here is built from what `Database` itself uses: the `Database` service, +// its open transaction, the client transaction when a client is in context, +// and the params the lookup was given. + +import { Cause, Context, Effect, Exit, MutableHashMap, Request, RequestResolver } from "effect" +import * as SqlClient from "effect/sql/SqlClient" +import type { NonEmptyArray } from "effect/Array" +import type { CompiledQueryDecodeError } from "../ch/compile" +import { QueryBuilderError } from "../ch/errors" +import type { ParamsSatisfied } from "../ch/expr" +import type { CHQuery, NeedsSelect } from "../ch/query" +import type { ColumnDefs } from "../ch/types" +import { Database, TransactionOpen } from "./database" +import type { DatabaseError } from "./errors" + +/** The errors of one batched lookup, besides `NoSuchElementError` for `findById`. */ +export type LookupError = DatabaseError | QueryBuilderError | CompiledQueryDecodeError + +/** + * Look up one row by its key. Concurrent calls in one batch become one + * statement. Call it directly when the query has no `param.*` markers; + * otherwise give the params first with `with`. + */ +export interface FindById { + (key: K & ParamsSatisfied): Effect.Effect + /** The lookup for these params. Lookups with equal params batch together. */ + readonly with: ( + params: Given & ParamsSatisfied, + ) => (key: K) => Effect.Effect +} + +/** + * Look up every row with a key. Concurrent calls in one batch become one + * statement. A key with no rows gives an empty array. + */ +export interface FindAllBy { + (key: K & ParamsSatisfied): Effect.Effect, LookupError, Database> + /** The lookup for these params. Lookups with equal params batch together. */ + readonly with: ( + params: Given & ParamsSatisfied, + ) => (key: K) => Effect.Effect, LookupError, Database> +} + +export interface LookupOptions { + /** The column to look up by. The query must select it under its own name. */ + readonly key: Key +} + +/** The params of a query, `never` when it has none. */ +type NoAny

= 0 extends 1 & P ? never : P + +interface LookupRequest extends Request.Request>, LookupError, Database> { + readonly key: unknown + readonly params: object +} + +const LookupRequest = Request.of() + +// Context objects get a number, so a batch key is a value `Equal` can compare +// without marking the objects themselves as compared by reference. +const ids = new WeakMap() +let nextId = 0 +const idOf = (value: object | undefined): number => { + if (value === undefined) return 0 + let id = ids.get(value) + if (id === undefined) { + id = ++nextId + ids.set(value, id) + } + return id +} + +const contextKey = (context: Context.Context, params: object): ReadonlyArray => { + const client = Context.getOrUndefined(context, SqlClient.SqlClient) + const connection = client === undefined ? undefined : Context.getOrUndefined(context, client.transactionService) + return [ + idOf(Context.getOrUndefined(context, Database)), + idOf(Context.getOrUndefined(context, TransactionOpen)), + idOf(connection), + params, + ] +} + +const resolverFor = (query: CHQuery, key: string) => { + // One stable key object per distinct batch key, dropped when its batch runs. + const keys = MutableHashMap.empty, object>() + return RequestResolver.makeWith({ + batchKey: (entry) => { + const value = contextKey(entry.context as Context.Context, entry.request.params) + const found = MutableHashMap.get(keys, value) + if (found._tag === "Some") return found.value + const stable = { value } + MutableHashMap.set(keys, value, stable) + return stable + }, + delay: Effect.yieldNow, + collectWhile: () => true, + runAll: (entries: NonEmptyArray>, batchKey) => + Effect.gen(function* () { + MutableHashMap.remove(keys, (batchKey as { value: ReadonlyArray }).value) + const state = query._state + if (state.limitValue !== undefined || state.offsetValue !== undefined) { + return yield* new QueryBuilderError({ + code: "InvalidArguments", + message: "a batched lookup cannot have limit or offset: they would apply to the whole batch, not to each key", + }) + } + const wanted = MutableHashMap.empty() + for (const entry of entries) MutableHashMap.set(wanted, entry.request.key, true) + const values = Array.from(MutableHashMap.keys(wanted)) + const db = yield* Database + const rows = yield* db.run( + query.where(($: any) => [$[key].in_(...values)]) as never, + entries[0].request.params as never, + ) + const byKey = MutableHashMap.empty>>() + for (const row of rows as ReadonlyArray>) { + const group = MutableHashMap.get(byKey, row[key]) + if (group._tag === "Some") group.value.push(row) + else MutableHashMap.set(byKey, row[key], [row]) + } + for (const entry of entries) { + const group = MutableHashMap.get(byKey, entry.request.key) + entry.completeUnsafe(Exit.succeed(group._tag === "Some" ? group.value : [])) + } + }).pipe(Effect.provideContext(entries[0].context)), + }) +} + +const lookup = ( + query: CHQuery, + key: string, + pick: (rows: ReadonlyArray>) => Effect.Effect, +) => { + const resolver = resolverFor(query, key) + const withParams = + (params: object) => + (value: unknown): Effect.Effect => + Effect.flatMap(Effect.request(LookupRequest({ key: value, params }), resolver), pick) + return Object.assign(withParams({}), { with: withParams }) +} + +/** + * A batched lookup of one row by `key`. Concurrent calls that share a batch + * (same `Database`, same transaction, same params) send one statement: + * `query` with `key IN (...)` added to its WHERE. Rows decode as `run` + * decodes them. A key with no row fails with `NoSuchElementError`; with + * several, the first one wins. + * + * Tenant scope comes from `query`: filter the tenant column there, with a + * value or a `param.*`, and `requireTenantScope` checks each batch statement. + * Lookups with different params never share a batch, so one statement never + * reads two tenants. + */ +export const findById = < + Cols extends ColumnDefs, + Output extends Record, + P, + const Key extends keyof Cols & keyof Output & string, +>( + query: CHQuery & NeedsSelect, + options: LookupOptions, +): FindById> => + lookup(query, options.key, (rows) => + rows.length === 0 ? Effect.fail(new Cause.NoSuchElementError()) : Effect.succeed(rows[0]), + ) as never + +/** + * A batched lookup of every row with `key`, such as the children of a parent. + * Batches as `findById` does. A key with no rows gives an empty array. + */ +export const findAllBy = < + Cols extends ColumnDefs, + Output extends Record, + P, + const Key extends keyof Cols & keyof Output & string, +>( + query: CHQuery & NeedsSelect, + options: LookupOptions, +): FindAllBy> => lookup(query, options.key, Effect.succeed) as never From c7c47681bf8932d46678a6e611188b10cbf5904c Mon Sep 17 00:00:00 2001 From: Taras Kornichuk Date: Sun, 25 Oct 2026 17:17:04 +0200 Subject: [PATCH 2/3] docs(database): document batched lookups --- docs/database.md | 73 +++++++++++++++++++++++++++++++++++++++++++++++ docs/reference.md | 1 + 2 files changed, 74 insertions(+) diff --git a/docs/database.md b/docs/database.md index 2e0cccd..c58fbf3 100644 --- a/docs/database.md +++ b/docs/database.md @@ -92,6 +92,8 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti | `retryContention(options?)` | Re-run on serialization failure or deadlock | | `Transaction` | The open transaction (`depth`, `isolationLevel`, `accessMode`) | | `isContention(error)` | Whether an error is a serialization failure or deadlock | +| `findById(query, { key })` | A batched lookup of one row by a key: N concurrent calls send one statement ([Batching lookups](#batching-lookups)) | +| `findAllBy(query, { key })` | A batched lookup of every row with a key, such as the children of a parent | `run`, `query`, `execute`, `transaction` and `retryContention` are also methods on `DatabaseApi`. @@ -239,6 +241,77 @@ transaction it fails with `TransactionOptionsRejected`, because the database has aborted the whole transaction. `transaction(body, { retry: "contention" })` is shorthand for `retryContention(transaction(body))`, so each attempt starts with a fresh BEGIN. +## Batching lookups + +Many fibers that each read one row by its id send one statement per fiber: the N+1 problem. +`findById` and `findAllBy` collect the lookups that run at the same time into one batch, with +Effect's `RequestResolver`, and send one statement for the batch: your query with +`key IN (...)` added to its WHERE. + +```ts title="batched-lookups.ts" +import { Effect } from "effect" +import * as Db from "@maple-dev/effect-orm/database" +import * as PG from "@maple-dev/effect-orm/postgres" + +const Users = PG.table("users", { + columns: { id: PG.int4, org_id: PG.text, name: PG.text }, + primaryKey: ["id"], + tenantColumn: "org_id", +}) +const Posts = PG.table("posts", { + columns: { id: PG.int4, org_id: PG.text, author_id: PG.int4, title: PG.text }, + primaryKey: ["id"], + tenantColumn: "org_id", +}) + +// Built once, for example at module level. The tenant is a param. +const userById = Db.findById( + PG.from(Users) + .select("id", "name") + .where(($) => [$.org_id.eq(PG.param.string("org"))]), + { key: "id" }, +) +const postsByAuthor = Db.findAllBy( + PG.from(Posts) + .select("id", "author_id", "title") + .where(($) => [$.org_id.eq(PG.param.string("org"))]) + .orderBy(["id", "asc"]), + { key: "author_id" }, +) + +// One SELECT for the users and one for their posts, whatever the number of ids. +export const authors = (org: string, ids: ReadonlyArray) => + Effect.forEach( + ids, + (id) => + Effect.all({ user: userById.with({ org })(id), posts: postsByAuthor.with({ org })(id) }, { concurrency: "unbounded" }), + { concurrency: "unbounded" }, + ) +``` + +- **Rows** decode as `run` decodes them. The query must select `key` under its own name, so + each row can go back to the lookup that asked for it. That is a type error if it does not. +- **Not found.** `findById` fails with `NoSuchElementError` for a key with no row; with more + than one row, the first one wins. `findAllBy` gives an empty array. +- **Params.** Call the lookup directly when the query has no `param.*` markers. Otherwise + give them first with `.with(params)`, which checks them as `run` does. +- **What shares a batch.** Lookups through the same `findById` that run at the same time, on + the same `Database`, in the same transaction (or both outside one), with equal params. A + lookup inside a transaction never runs on another connection, and one outside never runs + inside it. A key asked for twice is sent once. +- **Tenant scope.** Put the tenant filter in the query, with a value or a param. Lookups with + different params never share a batch, so one statement never reads two tenants, and + [`requireTenantScope`](./tenant-scoping.md#enforcing-it) checks each batch statement as it + checks `run`. +- **`limit` and `offset`** would apply to the whole batch, not to each key, so a query with + either fails with `QueryBuilderError` and sends nothing. + +The batch waits one `yieldNow`, the same as Effect's own resolvers, so the lookups must start +together: through `Effect.forEach` or `Effect.all` with concurrency, or in fibers forked in the +same step. Lookups in sequence send one statement each. + +_(Backed by `Request batching on PGlite`, in `src/database/resolver.test.ts`: one statement for N lookups, tenants and transactions kept apart.)_ + ## ClickHouse `clickhouseDialect` declares no transactions, so `transaction` fails with diff --git a/docs/reference.md b/docs/reference.md index abc2d46..0f7dc74 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -513,6 +513,7 @@ placeholders, double-quoted identifiers. | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors | `/migrate` | | `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `retryContention`, `Transaction` | `/database` | | `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | +| `findById`, `findAllBy`, `FindById`, `FindAllBy`, `LookupOptions`, `LookupError` — batched lookups by a key, one `IN` statement per batch | `/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` | From 8704891322fb5e66a8bc2798505676383295c3f1 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Mon, 2 Nov 2026 16:32:41 +0100 Subject: [PATCH 3/3] fix(database): split batched lookups to fit a dialect's bound-parameter limit D1 binds at most 100 values per statement, so a batch of more keys failed to compile. runAll now chunks the IN list to leave room for the query's own params, runs the chunks in turn on the batch's connection, and has D1/SQLite and live ClickHouse tests. Co-Authored-By: Claude Opus 5.5 --- .changeset/batched-lookups.md | 2 +- docs/database.md | 7 ++- src/database/resolver.sqlite.test.ts | 67 ++++++++++++++++++++++++ src/database/resolver.ts | 28 +++++++--- tests/clickhouse-http.clickhouse.test.ts | 15 ++++++ 5 files changed, 111 insertions(+), 8 deletions(-) create mode 100644 src/database/resolver.sqlite.test.ts diff --git a/.changeset/batched-lookups.md b/.changeset/batched-lookups.md index 44afe2a..ff4c586 100644 --- a/.changeset/batched-lookups.md +++ b/.changeset/batched-lookups.md @@ -2,4 +2,4 @@ "@maple-dev/effect-orm": minor --- -`Db.findById(query, { key })` and `Db.findAllBy(query, { key })` in `/database` batch lookups by a key with Effect's `RequestResolver`: N concurrent calls send one statement, the query with `key IN (...)` added to its WHERE, and each row decodes through the query's row schema. A batch never mixes `Database`s, transactions or params, so a tenant filter in the query (a value or a `param.*`, given with `.with(params)`) keeps each statement to one tenant, and `requireTenantScope` checks it. +`Db.findById(query, { key })` and `Db.findAllBy(query, { key })` in `/database` batch lookups by a key with Effect's `RequestResolver`: N concurrent calls send one statement, the query with `key IN (...)` added to its WHERE, and each row decodes through the query's row schema. A batch never mixes `Database`s, transactions or params, so a tenant filter in the query (a value or a `param.*`, given with `.with(params)`) keeps each statement to one tenant, and `requireTenantScope` checks it. On a dialect with a bound-parameter limit (D1: 100) a large batch is split into statements that each fit. diff --git a/docs/database.md b/docs/database.md index 6698e55..7fd84fc 100644 --- a/docs/database.md +++ b/docs/database.md @@ -345,6 +345,11 @@ export const authors = (org: string, ids: ReadonlyArray) => different params never share a batch, so one statement never reads two tenants, and [`requireTenantScope`](./tenant-scoping.md#enforcing-it) checks each batch statement as it checks `run`. +- **Bound-parameter limits.** On a dialect with a limit on the values one statement may + bind (D1: 100, Postgres: 65535), a large batch is split into statements that each fit, + with room for the query's own params. The statements run one after another on the + batch's connection. Each key is in exactly one of them, so its rows keep the query's + `ORDER BY`. ClickHouse inlines values and has no such limit. - **`limit` and `offset`** would apply to the whole batch, not to each key, so a query with either fails with `QueryBuilderError` and sends nothing. @@ -352,7 +357,7 @@ The batch waits one `yieldNow`, the same as Effect's own resolvers, so the looku together: through `Effect.forEach` or `Effect.all` with concurrency, or in fibers forked in the same step. Lookups in sequence send one statement each. -_(Backed by `Request batching on PGlite`, in `src/database/resolver.test.ts`: one statement for N lookups, tenants and transactions kept apart.)_ +_(Backed by `Request batching on PGlite`, in `src/database/resolver.test.ts`: one statement for N lookups, tenants and transactions kept apart; and `src/database/resolver.sqlite.test.ts`: a D1 batch over 100 bindings split into statements that fit.)_ ## ClickHouse diff --git a/src/database/resolver.sqlite.test.ts b/src/database/resolver.sqlite.test.ts new file mode 100644 index 0000000..bd05e98 --- /dev/null +++ b/src/database/resolver.sqlite.test.ts @@ -0,0 +1,67 @@ +// Batched lookups on SQLite and D1. D1 binds at most 100 values per statement, +// so a large batch goes in chunks that each fit, with room for the query's own +// params; SQLite's limit is far above any batch here, so it sends one. + +import * as D1Client from "@effect/sql-d1/D1Client" +import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient" +import { Effect, Layer } 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 Users = SQLite.table("users", { id: SQLite.integer, org: SQLite.text, name: SQLite.text }, { tenantColumn: "org" }) +const ids = Array.from({ length: 250 }, (_, i) => i + 1) + +const lookups = (dialect: Dialect, client: Layer.Layer) => + Effect.runPromise( + Effect.gen(function* () { + const statements: Array = [] + const db = Db.fromSqlClient(yield* SqlClient.SqlClient, { + dialect, + requireTenantScope: true, + observe: (statement) => Effect.sync(() => void statements.push(statement.sql)), + }) + yield* db.execute({ sql: "CREATE TABLE users (id INTEGER PRIMARY KEY, org TEXT NOT NULL, name TEXT NOT NULL)" }) + for (const id of ids) yield* db.execute({ sql: "INSERT INTO users VALUES (?, ?, ?)", parameters: [id, id % 2 === 0 ? "a" : "b", `u${id}`] }) + statements.length = 0 + const userById = Db.findById( + SQLite.from(Users) + .select("id", "name") + .where(($) => [$.org.eq(SQLite.param.string("org"))]), + { key: "id" }, + ) + const evens = ids.filter((id) => id % 2 === 0) + const found = yield* Effect.forEach(evens, (id) => userById.with({ org: "a" })(id), { concurrency: "unbounded" }).pipe( + Effect.provideService(Db.Database, db), + ) + return { found, statements } + }).pipe(Effect.provide(client)), + ) + +it("splits a D1 batch over the 100-binding limit into statements that each fit", async () => { + const mf = new Miniflare({ + modules: true, + script: "export default { fetch() { return new Response('resolver test') } }", + compatibilityDate: "2026-07-01", + d1Databases: ["DB"], + }) + try { + const binding = await mf.getD1Database("DB") + const { found, statements } = await lookups(SQLite.d1Dialect, D1Client.layer({ db: binding })) + expect(found).toEqual(ids.filter((id) => id % 2 === 0).map((id) => ({ id, name: `u${id}` }))) + // 125 keys plus the tenant param: 99 keys per statement. + expect(statements).toHaveLength(2) + for (const sql of statements) expect(sql.split("?").length - 1).toBeLessThanOrEqual(100) + } finally { + await mf.dispose() + } +}, 30_000) + +it("sends one statement for the same batch on SQLite", async () => { + const { found, statements } = await lookups(SQLite.sqliteDialect, SqliteClient.layer({ filename: ":memory:" })) + expect(found).toHaveLength(125) + expect(statements).toHaveLength(1) +}) diff --git a/src/database/resolver.ts b/src/database/resolver.ts index 40592dd..e6169f1 100644 --- a/src/database/resolver.ts +++ b/src/database/resolver.ts @@ -12,7 +12,7 @@ import { Cause, Context, Effect, Exit, MutableHashMap, Request, RequestResolver } from "effect" import * as SqlClient from "effect/sql/SqlClient" import type { NonEmptyArray } from "effect/Array" -import type { CompiledQueryDecodeError } from "../ch/compile" +import { compileCH, type CompiledQueryDecodeError } from "../ch/compile" import { QueryBuilderError } from "../ch/errors" import type { ParamsSatisfied } from "../ch/expr" import type { CHQuery, NeedsSelect } from "../ch/query" @@ -116,12 +116,28 @@ const resolverFor = (query: CHQuery, key: string) => { for (const entry of entries) MutableHashMap.set(wanted, entry.request.key, true) const values = Array.from(MutableHashMap.keys(wanted)) const db = yield* Database - const rows = yield* db.run( - query.where(($: any) => [$[key].in_(...values)]) as never, - entries[0].request.params as never, - ) + const params = entries[0].request.params as Record + // A dialect with a bound-parameter limit (D1: 100) would refuse one + // statement for a large batch, so the keys go in chunks that leave + // room for the params the query binds itself. Chunks run in turn, on + // the batch's own connection or transaction. Each key is in one + // chunk, so its rows keep the query's ORDER BY. + const style = db.dialect.params + let size = values.length + if (style._tag === "bind" && style.maxParameters !== undefined) { + const base = yield* compileCH(query as never, params as never, { dialect: db.dialect }) + // One key binds one value; with no room left the first chunk + // fails to compile with the limit's own error. + size = Math.max(1, style.maxParameters - base.parameters.length) + } + const rows: Array> = [] + for (let start = 0; start < values.length; start += size) { + const chunk = values.slice(start, start + size) + const sent = yield* db.run(query.where(($: any) => [$[key].in_(...chunk)]) as never, params as never) + for (const row of sent as ReadonlyArray>) rows.push(row) + } const byKey = MutableHashMap.empty>>() - for (const row of rows as ReadonlyArray>) { + for (const row of rows) { const group = MutableHashMap.get(byKey, row[key]) if (group._tag === "Some") group.value.push(row) else MutableHashMap.set(byKey, row[key], [row]) diff --git a/tests/clickhouse-http.clickhouse.test.ts b/tests/clickhouse-http.clickhouse.test.ts index 909cc34..9cf2957 100644 --- a/tests/clickhouse-http.clickhouse.test.ts +++ b/tests/clickhouse-http.clickhouse.test.ts @@ -26,6 +26,7 @@ const Live = Layer.effect( ).pipe(Layer.provideMerge(Client)) const Numbers = CH.table("system.numbers", { external: true, columns: { number: CH.uint64 } }) +const Databases = CH.table("system.databases", { external: true, columns: { name: CH.string, engine: CH.string } }) describe.skipIf(!endpoint)("ClickhouseHttpClient on ClickHouse", () => { it.layer(Live)((it) => { @@ -42,6 +43,20 @@ describe.skipIf(!endpoint)("ClickhouseHttpClient on ClickHouse", () => { }), ) + it.effect("batches concurrent findById lookups into one IN statement", () => + Effect.gen(function* () { + const byName = Db.findById(CH.from(Databases).select("name").crossTenant(), { key: "name" }) + const found = yield* Effect.forEach(["system", "default", "no_such_db"], (name) => Effect.result(byName(name)), { + concurrency: "unbounded", + }) + expect(found.map((result) => (result._tag === "Success" ? result.success : result.failure._tag))).toEqual([ + { name: "system" }, + { name: "default" }, + "NoSuchElementError", + ]) + }), + ) + it.effect("binds parameters on the server", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient