feat(database): Db.stream streams rows with decoding

.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.

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.

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` |

src/database.ts

@@ -17,6 +17,7 @@ export {
 	requireTransaction,
 	retryContention,
 	run,
+	stream,
 	transaction,
 	type DatabaseApi,
 	type RowOf,

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<Effect.Effect<ReadonlyA
 // @ts-expect-error the param is still required
 Db.run(upsert)
 expectTypeOf(PG.compileUnsafe(upsert, { count: 1 })).toEqualTypeOf<CH.CompiledQuery<{ readonly count: number }, undefined>>()
+
+// `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<Stream.Stream<Row, RunError, Db.Database>>()
+const byNote = CH.from(Items)
+	.select("id", "note")
+	.where(($) => [$.note.eq(CH.param.string("note"))])
+expectTypeOf(Db.stream(byNote, { note: "a" })).toEqualTypeOf<Stream.Stream<Row, RunError, Db.Database>>()
+// @ts-expect-error -- a missing param is a type error
+Db.stream(byNote)

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<Q> & ParamsSatisfied<QueryParams<Q>, Given>,
 		params?: Given,
 	) => Effect.Effect<ReadonlyArray<RowOf<Q>>, 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: <Q extends Runnable, const Given extends object = {}>(
+		query: Q & RunCheck<Q> & ParamsSatisfied<QueryParams<Q>, Given>,
+		params?: Given,
+	) => Stream.Stream<RowOf<Q>, DatabaseError | QueryBuilderError | CompiledQueryDecodeError>
 	/** Run a statement and return its rows, decoded through `schema` when given. */
 	readonly query: {
 		(statement: StatementInput): Effect.Effect<ReadonlyArray<Record<string, unknown>>, 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<Record<string, unknown>, DatabaseError> =>
+		cursorless
+			? Stream.fromIterableEffect(rows(statement))
+			: traceStream(
+					dialect,
+					statement,
+					Stream.unwrap(Effect.as(guard(statement), raw.unsafe<Record<string, unknown>>(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<any, any>,
+		wire: Stream.Stream<Record<string, unknown>, DatabaseError>,
+	): Stream.Stream<unknown, DatabaseError | CompiledQueryDecodeError> => {
+		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<never, DatabaseError | QueryBuilderError | CompiledQueryDecodeError>
+
 	const retryContention = <A, E, R>(effect: Effect.Effect<A, E, R>, retry?: RetryOptions) =>
 		Effect.flatMap(Effect.serviceOption(sql.transactionService), (open): Effect.Effect<A, E | TransactionOptionsRejected, R> =>
 			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 = <Q extends Runnable, const Given extends object = {}>(
 ): Effect.Effect<ReadonlyArray<RowOf<Q>>, DatabaseError | QueryBuilderError | CompiledQueryDecodeError, Database> =>
 	Effect.flatMap(Effect.service(Database), (db) => db.run(query, params))
 
+/** `stream` on the `Database` in context. */
+export const stream = <Q extends Runnable, const Given extends object = {}>(
+	query: Q & RunCheck<Q> & ParamsSatisfied<QueryParams<Q>, Given>,
+	params?: Given,
+): Stream.Stream<RowOf<Q>, 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<ReadonlyArray<Record<string, unknown>>, DatabaseError, Database>

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<number> = []
+			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<ReadonlyArray<number>> = []
+			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<Tracer.NativeSpan> = []
+			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" })
+		}),
+	)
+})