diff --git a/.changeset/db-stream.md b/.changeset/db-stream.md new file mode 100644 index 0000000..cf018c3 --- /dev/null +++ b/.changeset/db-stream.md @@ -0,0 +1,11 @@ +--- +"@maple-dev/effect-orm": minor +--- + +`Db.stream(query, params?)` and `DatabaseApi.stream` run what `run` takes and return a `Stream` +of its rows. Each row is decoded by the query's row schema when it arrives, so a large result +does not have to fit in memory. A row that does not match fails the stream with +`CompiledQueryDecodeError` and its `rowIndex`. The stream uses the `SqlClient` statement's +`.stream`, reads on the transaction's connection inside `transaction`, and runs in one client +span that counts the rows it emitted. On `sqliteDialect` and `d1Dialect`, whose Effect drivers +have no streaming read, the whole result is read first and then emitted row by row. diff --git a/docs/database.md b/docs/database.md index bf14d00..8888df9 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 | +| `stream(query, params?)` | `run` as a `Stream`: each row decoded as it arrives | | `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 | @@ -94,7 +95,7 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti | `Transaction` | The open transaction (`depth`, `isolationLevel`, `accessMode`) | | `isContention(error)` | Whether an error is a serialization failure or deadlock | -`run`, `query`, `execute`, `transaction` and `retryContention` are also methods on +`run`, `stream`, `query`, `execute`, `transaction` and `retryContention` are also methods on `DatabaseApi`. ### Queries and statements @@ -151,6 +152,29 @@ query compiled with the `/clickhouse` entry's `compile` would run on Postgres wi quoting and inlined params. Rows come back without the client's name transforms, because the decoder reads the aliases the compiler wrote. +### Streaming rows + +`stream` takes what `run` takes and returns a `Stream` of the same rows. Each row goes through +the query's row schema when it arrives, so an export or a reconciliation over a large table does +not hold the whole result in memory: + +```ts +const rows = Db.stream(PG.from(AccessLogs).select("id", "at", "lock_id").where(($) => [$.at.gte(since)])) +yield* rows.pipe(Stream.grouped(500), Stream.runForEach(writeBatch)) +``` + +The statement is sent when the stream runs, and again each time it runs. A row that does not +match the schema fails the stream with `CompiledQueryDecodeError` and its `rowIndex`, after the +rows before it; a failed statement fails it with `DatabaseError`. Inside `transaction`, the +stream reads on the transaction's connection, so run it to the end before the body returns. + +How rows arrive is the driver's: `stream` uses the `SqlClient` statement's `.stream`. The +ClickHouse clients read the response body as it comes. A driver with no streaming read, such as +PGlite, reads the whole result first and then emits it row by row. On `sqliteDialect` and +`d1Dialect` the result is always read whole first, because Effect's SQLite and D1 drivers have no +streaming read at all. A write streams its `RETURNING` rows, and one without `RETURNING` runs and emits nothing. +The statement's span stays open until the stream ends and counts the rows it emitted. + Build the `Database` wherever you build the client. If each request has its own pool, build a `Database` per request from it: nothing request-scoped is stored in the `Database`, and the open transaction lives in fiber context. diff --git a/docs/reference.md b/docs/reference.md index fe8f4eb..ebb67b6 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -512,7 +512,7 @@ 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` | +| `Database`, `run`, `stream`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` | | `DatabaseError`, `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` | diff --git a/src/database.ts b/src/database.ts index dc56aab..5695485 100644 --- a/src/database.ts +++ b/src/database.ts @@ -17,6 +17,7 @@ export { requireTransaction, retryContention, run, + stream, transaction, type DatabaseApi, type RowOf, diff --git a/src/database/database.test-d.ts b/src/database/database.test-d.ts index f684a71..739fceb 100644 --- a/src/database/database.test-d.ts +++ b/src/database/database.test-d.ts @@ -1,6 +1,6 @@ // Type-level tests: rows from `run`, `requireTransaction`, `afterCommit`, and what `transaction` removes. -import { Context, Effect, Schema } from "effect" +import { Context, Effect, Schema, Stream } from "effect" import { expectTypeOf } from "expect-type" import * as CH from "../ch/index" import * as PG from "../postgres" @@ -87,3 +87,12 @@ expectTypeOf(Db.run(upsert, { count: 1 })).toEqualTypeOf>() + +// `stream` infers the same rows as `run`, as a Stream, and checks params the same way. +expectTypeOf(Db.stream(CH.from(Items).select("id", "note"))).toEqualTypeOf>() +const byNote = CH.from(Items) + .select("id", "note") + .where(($) => [$.note.eq(CH.param.string("note"))]) +expectTypeOf(Db.stream(byNote, { note: "a" })).toEqualTypeOf>() +// @ts-expect-error -- a missing param is a type error +Db.stream(byNote) diff --git a/src/database/database.ts b/src/database/database.ts index 3a94eaa..5a0af6b 100644 --- a/src/database/database.ts +++ b/src/database/database.ts @@ -7,7 +7,7 @@ // statements outliving their transaction, and contention retry. See // `design/transactions.md`. -import { Cause, Context, Effect, ErrorReporter, Exit, Layer, Option, Schedule, Schema } from "effect" +import { Cause, Context, Effect, ErrorReporter, Exit, Layer, Option, Schedule, Schema, Stream } from "effect" import { dual } from "effect/Function" import * as SqlClient from "effect/sql/SqlClient" import { isSqlError } from "effect/sql/SqlError" @@ -31,7 +31,7 @@ import { } from "./errors" import { isSqlTemplate, renderTemplate, type SqlTemplate } from "./sql" import { firstLine, reasonOf, sqlStateOf, toDatabaseError } from "./sql-error" -import { traceStatement, type TracedStatement } from "./tracing" +import { traceStatement, traceStream, type TracedStatement } from "./tracing" /** SQL and the values bound to its placeholders. A `CompiledQuery` is one. */ export interface Statement { @@ -109,6 +109,17 @@ export interface DatabaseApi { query: Q & RunCheck & ParamsSatisfied, Given>, params?: Given, ) => Effect.Effect>, DatabaseError | QueryBuilderError | CompiledQueryDecodeError> + /** + * `run` as a stream: rows come out one at a time, each decoded by the + * query's row schema, so a large result never has to fit in memory. The + * statement is sent when the stream runs, and again each time it runs. + * Rows arrive as the driver reads them: a cursor where the driver has one, + * otherwise the whole result first. + */ + readonly stream: ( + query: Q & RunCheck & ParamsSatisfied, Given>, + params?: Given, + ) => Stream.Stream, DatabaseError | QueryBuilderError | CompiledQueryDecodeError> /** Run a statement and return its rows, decoded through `schema` when given. */ readonly query: { (statement: StatementInput): Effect.Effect>, DatabaseError> @@ -322,6 +333,56 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp : Effect.flatMap(rows(compiled), (wire) => compiled.decodeRows(wire)), ) + // Effect's SQLite drivers (node:sqlite, D1, …) have no streaming read: their + // `.stream` dies with "executeStream not implemented". A SQLite-family + // dialect reads the whole result as `run` does and emits it row by row. + const cursorless = dialect.functions === "sqlite" + + const wireStream = (statement: Statement & TracedStatement): Stream.Stream, DatabaseError> => + cursorless + ? Stream.fromIterableEffect(rows(statement)) + : traceStream( + dialect, + statement, + Stream.unwrap(Effect.as(guard(statement), raw.unsafe>(statement.sql, statement.parameters ?? []).stream)).pipe( + Stream.mapError(toDatabaseError(statement.sql)), + ), + ) + + // Each row through the query's row schema, as `decodeRows` does for a whole result. + const decodeStream = ( + compiled: CompiledQuery, + wire: Stream.Stream, DatabaseError>, + ): Stream.Stream => { + const schema = compiled.rowSchema + if (schema === undefined) return wire + const decode = Schema.decodeUnknownEffect(schema) + return wire.pipe( + Stream.zipWithIndex, + Stream.mapEffect(([row, rowIndex]) => + decode(row).pipe( + Effect.mapError( + (cause) => + new CompiledQueryDecodeError({ + message: `Compiled query row ${rowIndex} did not match its declared output schema`, + rowIndex, + cause, + }), + ), + ), + ), + ) + } + + const stream: DatabaseApi["stream"] = (runnable, params) => + Stream.unwrap( + Effect.map(Effect.flatMap(compileFor(runnable, params ?? {}), checkScope), (compiled) => + compiled.kind !== "select" && compiled.returning === undefined + ? Stream.fromEffectDrain(execute(compiled)) + : decodeStream(compiled, wireStream(compiled)), + ), + ) as Stream.Stream + const retryContention = (effect: Effect.Effect, retry?: RetryOptions) => Effect.flatMap(Effect.serviceOption(sql.transactionService), (open): Effect.Effect => Option.isSome(open) @@ -422,7 +483,7 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp ? once(body, options) : retryContention(once(body, options), options.retry === "contention" ? undefined : options.retry) - return { dialect, run, query, execute, transaction, retryContention } + return { dialect, run, stream, query, execute, transaction, retryContention } } /** @@ -498,6 +559,13 @@ export const run = ( ): Effect.Effect>, DatabaseError | QueryBuilderError | CompiledQueryDecodeError, Database> => Effect.flatMap(Effect.service(Database), (db) => db.run(query, params)) +/** `stream` on the `Database` in context. */ +export const stream = ( + query: Q & RunCheck & ParamsSatisfied, Given>, + params?: Given, +): Stream.Stream, DatabaseError | QueryBuilderError | CompiledQueryDecodeError, Database> => + Stream.unwrap(Effect.map(Effect.service(Database), (db) => db.stream(query, params))) + /** `query` on the `Database` in context. */ export const query: { (statement: StatementInput): Effect.Effect>, DatabaseError, Database> diff --git a/src/database/stream.test.ts b/src/database/stream.test.ts new file mode 100644 index 0000000..9d1eea9 --- /dev/null +++ b/src/database/stream.test.ts @@ -0,0 +1,169 @@ +// `Database.stream` on PGlite: rows come out one at a time, each decoded by +// the query's row Schema. PGlite has no cursor, so its driver reads the whole +// result and emits it as a stream; the contract checked here does not depend +// on that. + +import { PgliteClient } from "@effect/sql-pglite" +import { assert, expect, layer } from "@effect/vitest" +import { DateTime, Effect, Layer, Option, Stream, Tracer } from "effect" +import * as CH from "../ch/index" +import * as SqlClient from "effect/sql/SqlClient" +import * as PG from "../postgres" +import { postgresDialect } from "../pg/dialect" +import * as Db from "../database" + +const Live = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: postgresDialect }) + }), +).pipe(Layer.provideMerge(PgliteClient.layer({ postgresqlconf: "timezone = 'UTC'" }))) + +const Events = PG.table("stream_events", { + columns: { id: PG.int4, at: PG.timestamptz, note: PG.nullable(PG.text) }, + primaryKey: ["id"], +}) + +const at = (id: number) => DateTime.makeUnsafe(`2026-01-0${id}T00:00:00.000Z`) + +layer(Live, { excludeTestServices: true })("Database.stream on PGlite", (it) => { + it.effect("creates and fills the table", () => + Effect.gen(function* () { + yield* Db.execute(Db.sql`CREATE TABLE stream_events (id int4 PRIMARY KEY, at timestamptz NOT NULL, note text)`) + yield* Db.run( + PG.insertInto(Events).values([ + { id: 1, at: at(1), note: "a" }, + { id: 2, at: at(2), note: null }, + { id: 3, at: at(3), note: "c" }, + ]), + ) + }), + ) + + it.effect("streams the rows of a query, each decoded by its row schema", () => + Effect.gen(function* () { + const rows = yield* Stream.runCollect( + Db.stream(PG.from(Events).select("id", "at", "note").orderBy(["id", "asc"])), + ) + expect(rows).toEqual([ + { id: 1, at: at(1), note: "a" }, + { id: 2, at: at(2), note: null }, + { id: 3, at: at(3), note: "c" }, + ]) + expect(DateTime.isDateTime(rows[0]!.at)).toBe(true) + }), + ) + + it.effect("a row that does not match fails the stream with its index, after the rows before it", () => + Effect.gen(function* () { + yield* Db.execute(Db.sql`CREATE TABLE stream_flags (id int4 PRIMARY KEY, flag text)`) + yield* Db.execute(Db.sql`INSERT INTO stream_flags VALUES (1, NULL), (2, 'x'), (3, NULL)`) + // The table says boolean; the column holds text. + const Flags = PG.table("stream_flags", { columns: { id: PG.int4, flag: PG.nullable(PG.bool) }, primaryKey: ["id"] }) + const seen: Array = [] + const error = yield* Effect.flip( + Stream.runForEach(Db.stream(PG.from(Flags).select("id", "flag").orderBy(["id", "asc"])), (row) => + Effect.sync(() => void seen.push(row.id)), + ), + ) + expect(seen).toEqual([1]) + expect(error).toBeInstanceOf(CH.CompiledQueryDecodeError) + expect(error).toMatchObject({ rowIndex: 1 }) + }), + ) + + it.effect("runs on the transaction's connection inside a transaction", () => + Effect.gen(function* () { + const ids = Db.stream(PG.from(Events).select("id").orderBy(["id", "asc"])).pipe( + Stream.map((row) => row.id), + Stream.runCollect, + ) + // The body fails with what the stream saw, so the transaction rolls back. + const inside = yield* Db.transaction( + Db.run(PG.insertInto(Events).values({ id: 4, at: at(4), note: null })).pipe( + Effect.andThen(ids), + Effect.flatMap((seen) => Effect.fail(seen)), + ), + ).pipe(Effect.flip) + expect(inside).toEqual([1, 2, 3, 4]) + // The failure rolled the insert back. + expect(yield* ids).toEqual([1, 2, 3]) + }), + ) + + it.effect("an afterCommit hook queued beside a stream runs once the stream's transaction commits", () => + Effect.gen(function* () { + const ran: Array> = [] + const seen = yield* Db.transaction( + Effect.gen(function* () { + const ids = yield* Stream.runCollect(Db.stream(PG.from(Events).select("id").orderBy(["id", "asc"]))) + yield* Db.afterCommit(Effect.sync(() => void ran.push(ids.map((row) => row.id)))) + expect(ran).toEqual([]) + return ids.length + }), + ) + expect(seen).toBe(3) + expect(ran).toEqual([[1, 2, 3]]) + }), + ) + + it.effect("fills params, and streams the RETURNING rows of a write", () => + Effect.gen(function* () { + const byId = PG.from(Events) + .select("id", "note") + .where(($) => [$.id.eq(PG.param.int("id"))]) + expect(yield* Stream.runCollect(Db.stream(byId, { id: 3 }))).toEqual([{ id: 3, note: "c" }]) + const renamed = PG.update(Events) + .set({ note: "z" }) + .where(($) => [$.id.lte(2)]) + .returning(($) => ({ id: $.id, note: $.note })) + const rows = yield* Stream.runCollect(Db.stream(renamed)) + expect([...rows].sort((a, b) => a.id - b.id)).toEqual([ + { id: 1, note: "z" }, + { id: 2, note: "z" }, + ]) + // A write without RETURNING runs and emits nothing. + const reset = PG.update(Events).set({ note: null }).where(($) => [$.id.lte(2)]) + expect(yield* Stream.runCollect(Db.stream(reset))).toEqual([]) + expect(yield* Db.run(PG.from(Events).select("note").where(($) => [$.id.eq(1)]))).toEqual([{ note: null }]) + }), + ) + + it.effect("runs in one client span that counts the rows it emitted", () => + Effect.gen(function* () { + const spans: Array = [] + const recording = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan({ ...options, sampled: true }) + spans.push(span) + return span + }, + }) + const rows = yield* Stream.runCollect(Db.stream(PG.from(Events).select("id"))).pipe(Effect.withTracer(recording)) + expect(rows).toHaveLength(3) + const span = spans.find((s) => s.name === "SELECT stream_events") + assert(span !== undefined, `no statement span among ${spans.map((s) => s.name).join(", ")}`) + expect(span.kind).toBe("client") + expect(span.status._tag).toBe("Ended") + expect(Object.fromEntries(span.attributes)).toMatchObject({ + "db.system.name": "postgresql", + "db.operation.name": "SELECT", + "db.collection.name": "stream_events", + "db.response.returned_rows": 3, + }) + // Effect's own span runs inside it, carrying the query text. + const execute = spans.find((s) => s.attributes.has("effect.sql.method")) + assert(execute !== undefined) + expect(Option.getOrUndefined(execute.parent)).toBe(span) + }), + ) + + it.effect("a failed statement fails the stream with DatabaseError", () => + Effect.gen(function* () { + const Missing = PG.table("stream_missing", { columns: { id: PG.int4 }, primaryKey: ["id"] }) + const error = yield* Effect.flip(Stream.runCollect(Db.stream(PG.from(Missing).select("id")))) + expect(error).toBeInstanceOf(Db.DatabaseError) + expect(error).toMatchObject({ sqlState: "42P01" }) + }), + ) +}) diff --git a/src/database/tracing.ts b/src/database/tracing.ts index 5372ae8..2ca2f9c 100644 --- a/src/database/tracing.ts +++ b/src/database/tracing.ts @@ -5,7 +5,7 @@ // Effect's own span still runs inside it with the query text, but it cannot // carry these: it is named after the database, not the operation, and only a // compiled query knows its table. -import { Context, Effect } from "effect" +import { Context, Effect, Stream } from "effect" import type { CompiledQuery } from "../ch/compile" import type { Dialect } from "../ch/dialect" @@ -40,6 +40,32 @@ const OPERATIONS: Record, string> = { delete: "DELETE", } +// The span name and attributes for `statement`, given the summary in context. +const spanOf = (dialect: Dialect, statement: TracedStatement, configured: string | undefined) => { + const system = SYSTEM_NAMES[dialect.name] ?? dialect.name + const operation = statement.kind === undefined ? undefined : OPERATIONS[statement.kind] + const summary = + configured ?? (operation === undefined ? undefined : statement.collection === undefined ? operation : `${operation} ${statement.collection}`) + const attributes: Record = { "db.system.name": system } + if (operation !== undefined) attributes["db.operation.name"] = operation + if (statement.collection !== undefined) attributes["db.collection.name"] = statement.collection + if (summary !== undefined) attributes["db.query.summary"] = summary + return { name: summary ?? system, attributes } +} + +// `error.type` is the failure's class: a DatabaseError's reason, else its tag. +const annotateErrorType = (error: unknown): Effect.Effect => { + const reason = + typeof error === "object" && error !== null + ? "reason" in error && typeof error.reason === "string" + ? error.reason + : "_tag" in error && typeof error._tag === "string" + ? error._tag + : undefined + : undefined + return reason === undefined ? Effect.void : Effect.annotateCurrentSpan("error.type", reason) +} + /** * Runs `effect`, which sends `statement`, in its client span. The span name is * the summary when there is one, then `{operation} {table}`, then the system. @@ -52,31 +78,37 @@ export const traceStatement = ( rowsOf?: (result: A) => number, ): Effect.Effect => Effect.withFiber((fiber) => { - const configured = fiber.getRef(QuerySummary) - const system = SYSTEM_NAMES[dialect.name] ?? dialect.name - const operation = statement.kind === undefined ? undefined : OPERATIONS[statement.kind] - const summary = - configured ?? (operation === undefined ? undefined : statement.collection === undefined ? operation : `${operation} ${statement.collection}`) - const attributes: Record = { "db.system.name": system } - if (operation !== undefined) attributes["db.operation.name"] = operation - if (statement.collection !== undefined) attributes["db.collection.name"] = statement.collection - if (summary !== undefined) attributes["db.query.summary"] = summary + const { name, attributes } = spanOf(dialect, statement, fiber.getRef(QuerySummary)) return effect.pipe( Effect.tap((result) => rowsOf === undefined ? Effect.void : Effect.annotateCurrentSpan("db.response.returned_rows", rowsOf(result)), ), - // `error.type` is the failure's class: a DatabaseError's reason, else its tag. - Effect.tapError((error) => { - const reason = - typeof error === "object" && error !== null - ? "reason" in error && typeof error.reason === "string" - ? error.reason - : "_tag" in error && typeof error._tag === "string" - ? error._tag - : undefined - : undefined - return reason === undefined ? Effect.void : Effect.annotateCurrentSpan("error.type", reason) - }), - Effect.withSpan(summary ?? system, { kind: "client", attributes }), + Effect.tapError(annotateErrorType), + Effect.withSpan(name, { kind: "client", attributes }), ) }) + +/** + * `traceStatement` for a stream of rows: one client span from the first pull + * to the end, with `db.response.returned_rows` the rows it emitted. + */ +export const traceStream = (dialect: Dialect, statement: TracedStatement, stream: Stream.Stream): Stream.Stream => + Stream.unwrap( + Effect.withFiber((fiber) => { + const { name, attributes } = spanOf(dialect, statement, fiber.getRef(QuerySummary)) + // Counted per run: the effect `unwrap` takes runs each time the stream does. + let returned = 0 + return Effect.succeed( + stream.pipe( + Stream.tap(() => + Effect.sync(() => { + returned++ + }), + ), + Stream.onEnd(Effect.suspend(() => Effect.annotateCurrentSpan("db.response.returned_rows", returned))), + Stream.tapError((error: E) => annotateErrorType(error)), + Stream.withSpan(name, { kind: "client", attributes }), + ), + ) + }), + ) diff --git a/src/sqlite/d1.test.ts b/src/sqlite/d1.test.ts index c35df5a..c879d87 100644 --- a/src/sqlite/d1.test.ts +++ b/src/sqlite/d1.test.ts @@ -1,6 +1,6 @@ // A real local Workers D1 binding; no cloud account or network database needed. import * as D1Client from "@effect/sql-d1/D1Client" -import { Cause, Effect, Exit, Schema } from "effect" +import { Cause, Effect, Exit, Schema, Stream } from "effect" import * as SqlClient from "effect/sql/SqlClient" import { Miniflare } from "miniflare" import { expect, it } from "vitest" @@ -98,3 +98,14 @@ it("refuses a D1 integer past 2^53 instead of decoding a different number", asyn if (!Exit.isFailure(exit)) throw new Error(`an unsafe integer decoded as ${JSON.stringify(exit.value)}`) expect(Cause.findError(exit.cause)).toMatchObject({ success: { _tag: "@maple-dev/effect-orm/CompiledQueryDecodeError" } }) }, 30_000) + +it("streams rows on D1, whose driver has no streaming read, and keeps the tenant guard", async () => { + const result = await withD1((db) => Effect.gen(function* () { + yield* db.run(SQLite.insertInto(Ack).values([first, second, other]).crossTenant()) + const rows = yield* Stream.runCollect(db.stream(SQLite.from(Ack).select("id").where(($) => [$.org_id.eq("org_1")]).orderBy(["id", "asc"]))) + const unscoped = yield* Effect.exit(Stream.runCollect(db.stream(SQLite.from(Ack).select("id")))) + return { rows, unscoped } + })) + expect(result.rows).toEqual([{ id: "ack_1" }, { id: "ack_2" }]) + expect(Exit.isFailure(result.unscoped) && Cause.squash(result.unscoped.cause)).toBeInstanceOf(Db.TenantScopeViolation) +}) diff --git a/src/sqlite/sqlite.test.ts b/src/sqlite/sqlite.test.ts index fe5ff92..0f93463 100644 --- a/src/sqlite/sqlite.test.ts +++ b/src/sqlite/sqlite.test.ts @@ -1,5 +1,5 @@ import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient" -import { Cause, DateTime, Effect, Exit, Schema } from "effect" +import { Cause, DateTime, Effect, Exit, Schema, Stream } from "effect" import * as SqlClient from "effect/sql/SqlClient" import { expect, it } from "vitest" import * as Db from "../database" @@ -216,3 +216,11 @@ it("builds an external table from its columns, not from the definition's keys", expect(External.tenantColumn).toBe("org_id") expect(SQLite.compileUnsafe(SQLite.from(External).select("id").where(($) => [$.org_id.eq("o")])).sql).toContain('"ext"."org_id" = ?') }) + +it("streams rows on node:sqlite, whose driver has no streaming read, decoding each row", async () => { + const rows = await withDatabase((db) => Effect.gen(function* () { + yield* db.run(SQLite.insertInto(Items).values([1, 2].map((id) => ({ id, note: `${id}`, enabled: id === 1, score: null })))) + return yield* Stream.runCollect(db.stream(SQLite.from(Items).select("id", "enabled").orderBy(["id", "asc"]))) + })) + expect(rows).toEqual([{ id: 1, enabled: true }, { id: 2, enabled: false }]) +}) diff --git a/tests/clickhouse-http.clickhouse.test.ts b/tests/clickhouse-http.clickhouse.test.ts index 909cc34..4a4539b 100644 --- a/tests/clickhouse-http.clickhouse.test.ts +++ b/tests/clickhouse-http.clickhouse.test.ts @@ -58,6 +58,20 @@ describe.skipIf(!endpoint)("ClickhouseHttpClient on ClickHouse", () => { }), ) + it.effect("Db.stream decodes each streamed row with the query's row schema, and stops early", () => + Effect.gen(function* () { + const query = CH.from(Numbers).select(($) => ({ n: $.number })).where(($) => [$.number.lt(CH.param.int("limit"))]) + const rows = yield* Stream.runCollect(Db.stream(query, { limit: 3 })) + // UInt64 arrives as a string and is decoded to a number by the row schema. + expect(rows).toEqual([{ n: 0 }, { n: 1 }, { n: 2 }]) + // An unbounded scan ends when the consumer stops reading. + expect(yield* Stream.runCollect(Stream.take(Db.stream(CH.from(Numbers).select(($) => ({ n: $.number }))), 2))).toEqual([ + { n: 0 }, + { n: 1 }, + ]) + }), + ) + it.effect("classifies a syntax error", () => Effect.gen(function* () { const exit = yield* Effect.exit(Db.query({ sql: "SELEKT 1" })) diff --git a/tests/database.clickhouse.test.ts b/tests/database.clickhouse.test.ts index c43317c..60d0c3a 100644 --- a/tests/database.clickhouse.test.ts +++ b/tests/database.clickhouse.test.ts @@ -2,7 +2,7 @@ // refused before anything is sent. Creates one table in a throwaway database. import { ClickhouseClient } from "@effect/sql-clickhouse" -import { DateTime, Effect, Exit } from "effect" +import { DateTime, Effect, Exit, Stream } 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" @@ -198,6 +198,24 @@ describe("database", () => { expect(result.nulls).toEqual([{ Id: 1 }, { Id: 3 }]) }) + it("streams a query's rows through Effect's ClickHouse client, decoding each one", async () => { + const rows = await Effect.runPromise( + withDatabase((db) => + Effect.gen(function* () { + yield* db.execute(Db.sql`CREATE TABLE events (Id UInt64, Name String) ENGINE = MergeTree ORDER BY Id`) + yield* db.execute(Db.sql`INSERT INTO events VALUES (${1}, ${"a"}), (${2}, ${"b"}), (${3}, ${"c"})`) + const Events = CH.table("events", { external: true, columns: { Id: CH.uint64, Name: CH.string } }) + return yield* Stream.runCollect(db.stream(CH.from(Events).select("Id", "Name").orderBy(["Id", "asc"]))) + }), + ), + ) + expect(rows).toEqual([ + { Id: 1, Name: "a" }, + { Id: 2, Name: "b" }, + { Id: 3, Name: "c" }, + ]) + }) + it("refuses a transaction before sending anything", async () => { const result = await Effect.runPromise( withDatabase((db, sent) =>