diff --git a/.changeset/batched-lookups.md b/.changeset/batched-lookups.md new file mode 100644 index 0000000..ff4c586 --- /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. 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 bf14d00..7fd84fc 100644 --- a/docs/database.md +++ b/docs/database.md @@ -93,6 +93,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`. @@ -281,6 +283,82 @@ 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`. +- **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. + +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; and `src/database/resolver.sqlite.test.ts`: a D1 batch over 100 bindings split into statements that fit.)_ + ## ClickHouse `clickhouseDialect` declares no transactions, so `transaction` fails with diff --git a/docs/reference.md b/docs/reference.md index fe8f4eb..832914a 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -514,6 +514,7 @@ placeholders, double-quoted identifiers. | `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` | +| `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` | diff --git a/src/database.ts b/src/database.ts index dc56aab..f75453c 100644 --- a/src/database.ts +++ b/src/database.ts @@ -40,5 +40,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 3a94eaa..72b3175 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.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.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..e6169f1 --- /dev/null +++ b/src/database/resolver.ts @@ -0,0 +1,203 @@ +// 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 { compileCH, 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 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) { + 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 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