From ecace6a01db07626b0edf1d1e826d7f66fbadee4 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 4 Oct 2026 00:57:30 +0200 Subject: [PATCH] Add insertInto(...).values(...) for ClickHouse and Postgres INSERT ... VALUES built from the same table definitions queries read. The row type requires every column that is not nullable and has no default; TableOptions.defaults lists defaulted columns, and defineTable derives them and marks MATERIALIZED / ALIAS columns as not insertable. Values are encoded through each column's codec: inlined as literals on ClickHouse, bound as $n on Postgres. Columns are written in table order, and a key some rows omit becomes DEFAULT, which both ClickHouse matrix servers accept. Tenant scope is derived as for queries. compile and Database.run accept an insert. CompiledQuery.kind tells run to send an insert through command and return no rows. Postgres declares ParamStyle.maxParameters (65535); a statement over it fails to compile. Plan for the remaining write phases in design/writes.md. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 ++ README.md | 1 + design/writes.md | 241 ++++++++++++++++++++++++++++++ docs/README.md | 5 +- docs/database.md | 9 +- docs/inserts.md | 118 +++++++++++++++ docs/reference.md | 9 +- docs/tables-and-types.md | 4 + src/ch/compile.ts | 184 ++++++++++++++++++++++- src/ch/dialect.ts | 6 + src/ch/expr.ts | 2 +- src/ch/index.ts | 4 + src/ch/insert.test-d.ts | 92 ++++++++++++ src/ch/insert.test.ts | 189 +++++++++++++++++++++++ src/ch/insert.ts | 117 +++++++++++++++ src/ch/table.ts | 38 ++++- src/database/database.test.ts | 61 +++++++- src/database/database.ts | 18 ++- src/pg/dialect.ts | 2 + src/postgres.ts | 10 +- src/schema.ts | 2 + src/schema/define.ts | 59 +++++++- tests/database.clickhouse.test.ts | 51 ++++++- 23 files changed, 1186 insertions(+), 46 deletions(-) create mode 100644 design/writes.md create mode 100644 docs/inserts.md create mode 100644 src/ch/insert.test-d.ts create mode 100644 src/ch/insert.test.ts create mode 100644 src/ch/insert.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index af123ea..20067db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## Unreleased +- Add `insertInto(table).values(rows)` (see `docs/inserts.md`): INSERT ... VALUES from the same + table definitions, for ClickHouse and Postgres. The row type requires every column that is not + nullable and has no default; values are encoded through the column codecs, written as literals + on ClickHouse and bound on Postgres; a key some rows leave out is `DEFAULT`; tenant scope is + derived as for queries. `compile` and `Database.run` accept an insert; `run` sends it through + `command` and returns no rows. +- Add `TableOptions.defaults` for the columns an insert may leave out. `defineTable` works them + out from its column options and records `MATERIALIZED` / `ALIAS` columns as not insertable. +- Add `CompiledQuery.kind` (`"select"` or `"insert"`); `rawCompiledQuery` takes it as an option. +- Add `ParamStyle.maxParameters`; Postgres sets 65535, and a statement over it fails to compile. - Add `@maple-dev/effect-orm/database`, opt-in (see `docs/database.md`): a `Database` over the Effect `SqlClient` you already use. `run(query, params?)` compiles a query for the database's dialect, runs it and decodes its rows; `sql\`...\`` writes the other statements with every diff --git a/README.md b/README.md index 419acec..5b316f8 100644 --- a/README.md +++ b/README.md @@ -149,6 +149,7 @@ Full guides live in [`docs/`](./docs/README.md): | [Expressions and conditions](./docs/expressions.md) | Comparisons, arithmetic, optional predicates, aggregates | | [Joins and subqueries](./docs/joins-and-subqueries.md) | The join family, `fromQuery`, correlated subqueries | | [Unions and CTEs](./docs/unions-and-ctes.md) | `unionAll`, `fromUnion`, `withCTE` | +| [Inserting rows](./docs/inserts.md) | `insertInto`, the insert row type, `DEFAULT`, binding | | [Params and compilation](./docs/params-and-compilation.md) | `param.*`, how values reach the SQL, `CompiledQuery` | | [Decoding results](./docs/decoding-results.md) | `rowSchema`, `decodeRows`, decode errors | | [Running a query](./docs/running-queries.md) | Executing the SQL with a real client, wire settings, `SETTINGS` | diff --git a/design/writes.md b/design/writes.md new file mode 100644 index 0000000..7fd080b --- /dev/null +++ b/design/writes.md @@ -0,0 +1,241 @@ +# Writes: INSERT + +Status: phase 1 built (`insertInto(...).values(...)`, see `docs/inserts.md`); phases 2 to 4 not +started. Section 11 lists where the build differs from the plan. UPDATE and DELETE come later and +will reuse what this note sets up (the write-statement state, the `RETURNING` path, value +encoding). + +## Goal + +Build INSERT statements from the same table definitions the SELECT builder reads, for both +dialects, so that the ~101 Maple insert sites (`design/transactions.md` §5) and ClickHouse +ingest code stop writing SQL by hand. Writes meet the interface §5 of the transactions plan +already fixed: **a write compiles to a `CompiledQuery` whose `decodeRows` decodes the +`RETURNING` list (an empty row schema without one), so `Database.run` runs writes and reads +alike.** + +Same rules as the reads: the builder is an immutable value, nothing touches the network, values +are encoded through the column's own codec (the same one that decodes it), and a clause a +dialect lacks fails at compile time with a `QueryBuilderError` instead of producing SQL that +"may run". + +## What it looks like + +```ts +// Postgres +const insertKey = CH.insertInto(ApiKeys) + .values({ id: CH.param.string("id"), orgId: CH.param.string("orgId"), name: "default" }) + .returning(($) => ({ id: $.id, createdAt: $.createdAt })) + +const [row] = yield* Database.run(insertKey, { id, orgId }) // { id: string; createdAt: DateTime.Utc } + +// multi-row, values inline +yield* Database.run(CH.insertInto(Events).values(rows)) // rows: ReadonlyArray> + +// INSERT ... SELECT, output checked against the target's columns +CH.insertInto(DailyRollup).select(CH.from(Traces).select(($) => ({ day: ..., OrgId: $.OrgId, count: CH.count() }))) + +// upsert (Postgres) +CH.insertInto(Counters) + .values({ key: "k", count: 1 }) + .onConflict(["key"]) + .doUpdate(($, excluded) => ({ count: $.count.add(excluded.count) }), { where: ($) => $.locked.eq(false) }) +``` + +## 1. Builder and state + +New file `src/ch/insert.ts`, beside `query.ts`: + +```ts +interface CHInsertState { + readonly table: Table + readonly source: + | { readonly _tag: "Values"; readonly rows: ReadonlyArray> } + | { readonly _tag: "Select"; readonly query: CHQuery | CHUnionQuery } + readonly returningFn?: ($: ColumnAccessor) => Record> + readonly conflict?: ConflictClause // §5 + readonly settings?: Readonly> // ClickHouse, §6 +} + +interface CHInsert { + readonly _tag: "CHInsert" + readonly _state: CHInsertState + values(row: InsertRow | ReadonlyArray>): CHInsert<...> + select(query: Q & FitsTarget): CHInsert<...> + returning(fn: ($: ColumnAccessor) => S): CHInsert, Route> + onConflict(...), settings(...), route(...) +} +``` + +`insertInto(table)` returns a `CHInsert` with no source; compiling one without `values` or +`select` is a `QueryBuilderError` (`code: "EmptyInsert"`), as is `values([])`. + +## 2. The insert row type + +```ts +type InsertRow> = + { readonly [K in Exclude]: InsertValue } & + { readonly [K in Optional]?: InsertValue } + +type InsertValue = Comparable> | Expr>> // a value, a param, or an expression +``` + +- A value is the decoded TS type (`DateTime.Utc`, also `Date`/string where `Comparable` already + accepts them), a `param.*` marker, or any `Expr` of the column's type (`CH.now()`, + `rawExpr(...)`). Params make an insert a reusable prepared value, like a SELECT. +- `undefined` (or an absent key) means "use the column's default"; `null` means NULL and only + type-checks on a `nullable(...)` column. +- **Which columns are optional**: nullable columns, plus columns that declare a default. + `defineTable` already knows this (`ColumnOptions.default` / `defaultExpr`), but `ColumnsOf` + throws it away; it must keep a phantom set of defaulted keys on `SchemaTable`. `MATERIALIZED` + and `ALIAS` columns are `Computed`: not insertable at all, a type error if present. +- Plain `table()` has no defaults information today. Add `TableOptions.defaults?: ReadonlyArray` + (Postgres `serial` / `DEFAULT now()` columns, which the query-side `table()` is what Maple + uses until Postgres `defineTable` exists). Open question 1. + +## 3. Compiling VALUES + +New `src/ch/compile-insert.ts`. `compileCH` / `compile` dispatch on `_tag: "CHInsert"`, so every +existing entry point (root `compile`, `postgres` `compile`, `Database.run`) accepts an insert +without a new name. + +- **Column list**: the union of keys over all rows, in table-definition order (not object + order, so two rows with different key order cannot swap values, the same rule unions follow). + Always written out: `INSERT INTO t (a, b) VALUES ...`, never positional. +- **A key missing from some rows**: Postgres writes `DEFAULT` in that slot. ClickHouse: verify + on the matrix whether `DEFAULT` is accepted in `VALUES`; if not, fail with + `code: "RaggedInsert"` and tell the caller to split the batch. Open question 2. +- **Values**: each literal is encoded through the column's codec (`encodeColumnLiteral` + already does this for DDL defaults). ClickHouse inlines it as a literal; Postgres binds it as + `$n`, with the column's SQL type as a cast (`$3::timestamptz`) so an untyped param cannot be + inferred wrong — the problem `placeholderCasts` solves for params today, but here the column + type is known exactly. Params and expressions go through the existing fragment renderer and + `renderParams`, so one statement mixes them freely. +- **Bind limit**: Postgres caps a statement at 65535 parameters. Exceeding it is a + `QueryBuilderError` (`code: "TooManyParameters"`) naming the row count that would fit. No + automatic chunking: it would silently split one atomic statement into several and reorder + `RETURNING` across them. +- **Row schema**: none without `returning` (`rowSchemaSource: "derived"`, empty struct, so + `decodeRows` over the empty result is exact rather than an identity cast). + +### Tenant scope + +An insert has a scope like a SELECT, derived, never asserted: + +- target table has a `tenantColumn`: the type already requires it. `single-tenant` when every row + gives it the same param or the same literal; `cross-tenant` when rows differ; an `Expr` that + is not a param or literal makes it `cross-tenant` (the proof cannot see through it). +- no tenant column: `untenanted`. +- `INSERT ... SELECT`: the SELECT's scope, combined with the rule above for the selected tenant + column. + +## 4. `RETURNING` and running writes + +- `Dialect.clauses.returning: boolean` (Postgres true, ClickHouse false). `returning` on + ClickHouse fails at compile, like `.format()` on Postgres. +- The row schema is derived from the returning select exactly as `deriveRowSchema` does for + SELECT, so `Database.run(insert)` returns typed, decoded rows. `RowOf>` is `R`. +- **ClickHouse execution**: `Database.run` sends queries through the client's query path, which + asks for a JSON result an INSERT does not have. A compiled write without `RETURNING` must go + through the same `command` wrapper `execute` uses. `CompiledQuery` gains + `kind: "select" | "write"` so `run` can choose; `run` returns `[]` for a write without + `RETURNING`. +- `Runnable` widens to include `CHInsert`; `compileFor` gains the third branch. + +## 5. `INSERT ... SELECT` + +- Typed: the SELECT's output must fit the target columns, checked with the `MisfitColumns` type + `materializedView` already uses (move it out of `schema/define.ts` into `ch/types.ts`). A + required target column missing from the output is also a type error. +- The column list is the output's aliases in select order, so ClickHouse's positional matching + and Postgres's agree. +- ClickHouse backfills over big tables can outlast an HTTP timeout (`design/migrations.md`); that + stays the caller's concern, but the docs page says so. + +## 6. `ON CONFLICT` (Postgres) + +Maple: 25 `DO UPDATE`, 43 `DO NOTHING`, 66 `excluded.` references, 6 `setWhere`. + +```ts +.onConflict(target) // ["col", ...] | { constraint: "name" } | omitted (DO NOTHING only) + .doNothing() + .doUpdate(($, excluded) => ({ col: expr, ... }), { where?: ($, excluded) => Condition }) +``` + +- `excluded` is a `ColumnAccessor` over the target's columns rendering as `excluded."col"`, so + `excluded.count` is an `Expr` of the column's type. +- The `set` record is typed like an insert row (values, params, expressions), all keys optional. +- `Dialect.clauses.onConflict: boolean`; ClickHouse fails at compile and the message points at + `ReplacingMergeTree`. +- `DO NOTHING` with `RETURNING` returns no row for a skipped insert; the docs say so, because it + is the usual Drizzle surprise. + +## 7. ClickHouse specifics + +- `.settings({ async_insert: 1, wait_for_async_insert: 1 })` renders + `INSERT INTO t (...) SETTINGS ... VALUES ...`. Keys are checked as identifiers, values as + literals. Postgres fails at compile. Note: an insert inside a ClickHouse transaction needs + `async_insert=0` (`design/transactions.md` §3), irrelevant while ClickHouse transactions are + `none`. +- **Bulk ingest**: large batches should not be SQL text. Add `encodeInsertRows(table, rows)` + returning wire-shaped JSON objects (the column codecs run backwards, the way `encodeRows` + works) for `client.insert({ format: "JSONEachRow" })`. Same row type and validation as + `values`, no SQL. Phase 4; skipped if no consumer wants it. + +## 8. Tests + +- **Exact SQL** (`src/ch/insert.test.ts`, snapshot in `tests/core-sql.test.ts`): single and + multi-row, column order independent of key order, defaults omitted, `DEFAULT` slots, NULL, + params, expressions, every column type's literal for both dialects, `TooManyParameters`, + `EmptyInsert`, unsupported clauses per dialect, tenant scope cases. +- **Types** (`src/ch/insert.test-d.ts`): required vs optional keys, `null` only on nullable, + `MATERIALIZED` / `ALIAS` rejected, branded columns accept plain params, `returning` infers + `RowOf`, `INSERT ... SELECT` misfit and missing-column errors. +- **PGlite** (`src/pg/postgres.test.ts`): insert then select round-trips every type; + `RETURNING` decodes; `ON CONFLICT` both forms with `excluded` and `where`; insert inside + `Database.transaction` rolls back. +- **ClickHouse matrix** (`tests/*.clickhouse.test.ts`): `Database.run(insert)` goes through + the command path on every server; round-trip of every column type incl. `Map`, `Array`, + `Nullable`, `DateTime64`; `DEFAULT` in `VALUES` (answers open question 2); `SETTINGS`; + `INSERT ... SELECT`. +- **Docs**: `docs/inserts.md` with examples extracted by `check-doc-examples.mjs`, citations to + `src/docs-examples.test.ts`; update `docs/database.md` ("the builder compiles SELECTs only"), + `docs/README.md` ("does not ... insert rows"), `docs/reference.md`, and the exports check. + +## 9. Phases + +1. **VALUES on both dialects.** `insertInto`, `values`, the row type with defaults (§2), + compile dispatch, value encoding and binding, tenant scope, `CompiledQuery.kind`, + `Database.run` command path. Unblocks most Maple sites that do not read back. +2. **`RETURNING`.** `Dialect.clauses.returning`, derived row schema, `RowOf`. +3. **`ON CONFLICT`.** `doNothing`, `doUpdate` with `excluded` and `where`. +4. **`INSERT ... SELECT`** and ClickHouse `SETTINGS`; `encodeInsertRows` if wanted. + +Each phase lands with its tests and docs. UPDATE and DELETE follow as their own phases in this +note, reusing `kind: "write"`, the returning path and the `set` record type from §6. + +## 10. Decisions + +1. **Defaults on plain `table()`**: `TableOptions.defaults`, a list of column names. + `defineTable` derives the same from `default` / `defaultExpr`, and records `materialized` / + `alias` columns as computed. `column()` infers which option names were given, so this needs + no change at its call sites. +2. **Ragged multi-row batches on ClickHouse**: `DEFAULT` in `VALUES` works on both matrix + servers (26.2.19.43 and 26.8.2.7), including for an expression default, so both dialects + write `DEFAULT`. +3. **One name**: `compile` / `compileUnsafe` are overloaded for an insert; `Database.run` takes + one unchanged. +4. **Tenant scope for writes**: derived and reported as for reads; nothing is refused. + +## 11. Where the build differs + +- **Values are bound without a cast.** §3 planned `$3::timestamptz`. Postgres coerces a + placeholder in `INSERT ... VALUES` to the target column already, so the cast adds nothing. + Each value is encoded through its column's `literalSchema` (so a failure names the row and + column), then passed as an internal param of an identity type: ClickHouse inlines it, + Postgres binds it, through the same `renderParams` every query uses. +- **`CompiledQuery.kind`** is `"select" | "insert"`, not `"select" | "write"`. `rawCompiledQuery` + takes it as an option, so a handwritten INSERT can say so too. +- **The bound-value limit is a dialect field**, `ParamStyle.maxParameters`, so it applies to + queries as well and to a later dialect. The error is `InvalidArguments`; no new error code. +- **An insert's row schema** is an empty struct with `rowSchemaSource: "derived"`. diff --git a/docs/README.md b/docs/README.md index 0e2dba6..1915615 100644 --- a/docs/README.md +++ b/docs/README.md @@ -15,8 +15,8 @@ well as an Effect application. The database client remains your choice. You do not need a Maple account, Maple's schema, or tenant columns. Tenant analysis is an optional feature for applications that share tables between tenants. -The root builder does not manage connections, create tables, run migrations, insert rows, or provide -an ORM. Opt-in [schema and migration entry points](./migrations.md) add DDL and migrations for +The root builder does not manage connections, create tables, or run migrations. It builds +SELECTs and [INSERTs](./inserts.md); UPDATE and DELETE are not built yet. Opt-in [schema and migration entry points](./migrations.md) add DDL and migrations for ClickHouse. It does not validate SQL against a live server, choose query plans, enforce authorization, or supply retries. Existing ClickHouse tables and your executor own those responsibilities. [Getting started](./getting-started.md) covers npm installation and building from source. @@ -46,6 +46,7 @@ Roughly in reading order. | [Expressions and conditions](./expressions.md) | Comparisons, arithmetic, optional predicates, aggregates | | [Joins and subqueries](./joins-and-subqueries.md) | The join family, `fromQuery`, correlated subqueries | | [Unions and CTEs](./unions-and-ctes.md) | `unionAll`, `fromUnion`, `withCTE` | +| [Inserting rows](./inserts.md) | `insertInto`, the insert row type, `DEFAULT`, binding | | [Params and compilation](./params-and-compilation.md) | `param.*`, how values reach the SQL, `CompiledQuery` | | [Decoding results](./decoding-results.md) | `rowSchema`, `decodeRows`, `decodeFirstRow`, decode errors | | [Running a query](./running-queries.md) | Executing the SQL with a real client, wire settings, `SETTINGS` | diff --git a/docs/database.md b/docs/database.md index d21a857..5401a8a 100644 --- a/docs/database.md +++ b/docs/database.md @@ -99,12 +99,12 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti ### Queries and statements -`run` takes the query you built, a `unionAll`, or a query compiled elsewhere. It compiles with +`run` takes the query you built, a `unionAll`, an `insertInto`, or a query compiled elsewhere. It compiles with the database's dialect, so you never pick a `compile`; `params` fills the query's `param.*` markers, and a missing one fails with `QueryBuilderError`. A query compiled elsewhere must have been compiled for the same dialect, or `run` dies: the root `compile` is ClickHouse's. -`sql` writes the statements the builder does not have yet (INSERT, UPDATE, DDL, advisory +`sql` writes the statements the builder does not have yet (UPDATE, DELETE, DDL, advisory locks). Each `${value}` is bound, as `$1, $2, ...` on Postgres and as an escaped literal on ClickHouse, so nothing in a value becomes SQL. A `sql` inside another is spliced, so statements compose. Names go through `sql.identifier`, which accepts only plain identifiers @@ -247,5 +247,6 @@ fails at BEGIN today: through the query path with a syntax error, and through `a ## Writes -The builder compiles SELECTs only, so far. Write INSERT, UPDATE and DELETE with `sql`, as -above, and read `RETURNING` with `query` and a schema. +The builder compiles SELECTs and [INSERTs](./inserts.md). `run` runs an insert and returns no +rows; on ClickHouse it goes through `command`, as `execute` does. Write UPDATE, DELETE and +`INSERT ... RETURNING` with `sql`, as above, and read `RETURNING` with `query` and a schema. diff --git a/docs/inserts.md b/docs/inserts.md new file mode 100644 index 0000000..29f0baa --- /dev/null +++ b/docs/inserts.md @@ -0,0 +1,118 @@ +# Inserting rows + +`insertInto(table).values(rows)` builds an INSERT from the same table definition your queries +read. Like a query, it is an immutable value: nothing is sent until you run it, and `compile` +writes it for the dialect you compile with. + +```ts +import * as CH from "@maple-dev/effect-orm" +import * as Db from "@maple-dev/effect-orm/database" +import * as PG from "@maple-dev/effect-orm/postgres" + +const ApiKeys = CH.table( + "api_keys", + { + id: PG.uuid, + org_id: PG.text, + name: PG.text, + created_at: PG.timestamptz, + revoked: PG.bool, + note: PG.nullable(PG.text), + }, + { tenantColumn: "org_id", defaults: ["created_at", "revoked"] }, +) + +const insertKey = CH.insertInto(ApiKeys).values({ + id: CH.param.string("id"), + org_id: CH.param.string("orgId"), + name: "default", +}) + +// yield* Db.run(insertKey, { id, orgId }) +``` + +`Database.run` compiles the insert for its database's dialect and runs it. An insert returns no +rows; `RETURNING` is not built yet (see [`design/writes.md`](../design/writes.md)). + +## The row type + +Each row is typed from the table: + +- A column is **required** unless it is nullable or listed in `defaults`. +- Leaving an optional column out, or passing `undefined`, writes the column's default. +- `null` writes NULL, and only type-checks on a nullable column. +- A value can be a plain value of the column's type, a `param.*` of it, or any expression of it, + such as `CH.rawExpr("now()", CH.dateTime)`. A `DateTime` column also takes a `Date` or the + `'YYYY-MM-DD hh:mm:ss'` string, as in a comparison. + +`InsertRowOf` names the row type, for a function that builds rows. + +### Which columns have defaults + +`table()` cannot see your DDL, so you list the columns the database fills in with +`defaults`: a Postgres `serial` or `DEFAULT now()`, a ClickHouse `DEFAULT`. A table declared +with [`defineTable`](./migrations.md) works this out from its column options: a column with +`default` or `defaultExpr` is optional, and a `materialized` or `alias` column cannot be inserted +at all (it is not in the row type, and a row that names it anyway fails to compile). + +ClickHouse fills every column it is not given with a default, even without a `DEFAULT` clause: +`0` for a number, `''` for a string. The row type still requires those columns unless you list +them, so a forgotten value is a type error rather than a silent zero. + +## What it compiles to + +Columns are written in table order, whatever order the keys are in, so two rows with their keys +in different orders cannot swap values. A column that some rows give and others leave out is +`DEFAULT` in the rows that leave it out: + +```ts +const Events = CH.table( + "events", + { OrgId: CH.string, Id: CH.uint64, At: CH.dateTime }, + { tenantColumn: "OrgId", defaults: ["Id"] }, +) + +CH.compileUnsafe( + CH.insertInto(Events).values([ + { At: new Date(0), OrgId: "o1" }, + { OrgId: "o1", Id: 5, At: "2026-01-01 00:00:00" }, + ]), +).sql +// INSERT INTO events (OrgId, Id, At) +// VALUES ('o1', DEFAULT, '1970-01-01 00:00:00'), ('o1', 5, '2026-01-01 00:00:00') +``` + +Every value is encoded through its column's codec, the same one that decodes the column. A value +the codec rejects fails to compile with a `QueryBuilderError` that names the row and column. + +- **ClickHouse** writes each value into the SQL as an escaped literal, as it does for params. +- **Postgres** binds each value as `$1, $2, ...` and returns them in `parameters`. A param used in + several rows is bound once. A statement over 65535 bound values (Postgres's limit) fails to + compile instead of being split: send fewer rows per statement. + +## Tenant scope + +An insert has a `tenantScope` like a query, worked out the same way. On a table with a +`tenantColumn`, it is `"single-tenant"` when every row gives that column the same value or the +same param, and `"cross-tenant"` when rows differ or a row uses another expression. A table +without a tenant column gives `"untenanted"`. + +## Failures + +| Case | Result | +| ------------------------------------------------- | ---------------------------------------- | +| `values([])`, a row with no values | `QueryBuilderError` `InvalidArguments` | +| A key that is not a column, or a computed column | `QueryBuilderError` `InvalidArguments` | +| A value the column's codec rejects | `QueryBuilderError` `InvalidLiteral` | +| A param with no value | `QueryBuilderError` `UnresolvedParam` | +| Over the dialect's bound-value limit | `QueryBuilderError` `InvalidArguments` | +| Compiling without `values` | `QueryBuilderDefect` | + +_(Backed by `src/ch/insert.test.ts`, `src/database/database.test.ts` and +`tests/database.clickhouse.test.ts`.)_ + +## Large batches on ClickHouse + +An insert is SQL text, which suits the batches an application writes per request. For bulk +ingest, prefer your client's own insert with `JSONEachRow`, which streams rows instead of +building one statement. diff --git a/docs/reference.md b/docs/reference.md index 1e145d7..5055b6f 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -62,6 +62,7 @@ Note `/sql` exports a `compile` (fragment → string) distinct from the root `co | `fromQuery` | `(query, alias) => CHQuery` | | `fromUnion` | `(union, alias) => CHQuery` | | `unionAll` | `(...queries) => CHUnionQuery` | +| `insertInto` | `(table) => CHInsert`; `.values(row \| rows)` sets its rows. See [Inserting rows](./inserts.md) | ### `CHQuery` methods @@ -86,11 +87,11 @@ Note `/sql` exports a `compile` (fragment → string) distinct from the root `co | Export | Signature | | -------------------- | ------------------------------------------------------------------------------------ | -| `compile` | `(query, params, options?) => Effect, QueryBuilderError>` | +| `compile` | `(query, params, options?) => Effect, QueryBuilderError>`; also `(insert, params?, options?)`, whose options (`InsertCompileOptions`) are only `dialect` | | `compileUnsafe` | The same, returning `CompiledQuery` and throwing instead | | `compileUnion` | `(union, params, options?) => Effect, QueryBuilderError>` | | `compileUnionUnsafe` | The same, throwing instead | -| `rawCompiledQuery` | `({ sql, tenantScope, reason, justification, rowSchema?, route?, dialect? }) => CompiledQuery` | +| `rawCompiledQuery` | `({ sql, tenantScope, reason, justification, rowSchema?, route?, dialect?, kind? }) => CompiledQuery` | | `clickhouseDialect` | The default `Dialect`: params written into the SQL as ClickHouse literals | `Dialect`, `DialectClauses` and `ParamStyle` describe a database: how identifiers and literals @@ -278,7 +279,7 @@ Types: `WindowSpec`, `CompiledWindowSpec`, `WindowFrameBound`, `WindowRowsFrame` **Everything else** — `Table`, `TableOptions`, `Expr`, `ColumnRef`, `Condition`, `Comparable` (what a value of a type may be compared against), `MapValueOf`, `Subquery`, `ParamMarker`, -`ParamKind`, `CHQuery`, `CHUnionQuery`, `ColumnAccessor`, `JoinedColumnAccessor`, +`ParamKind`, `CHQuery`, `CHUnionQuery`, `CHInsert`, `InsertRow`, `InsertRowOf`, `InsertValue`, `ColumnAccessor`, `JoinedColumnAccessor`, `JoinOnCallback`, `CompiledQuery`, `CompiledQueryInput`, `CompiledQueryRowSchema`, `RowSchemaMismatch`, `TenantScope`, `Dialect`, `DialectClauses`, `DialectTransactions`, `IsolationLevel`, `TransactionSettings`, `ParamStyle`, `FnResult`, `WindowFunnelMode`, `WindowSpec`, `WindowRowsFrame`, `WindowFrameBound`, `WindowOrderDirection`, `CompiledWindowSpec`. @@ -297,7 +298,7 @@ Tag `"@maple-dev/effect-orm/QueryBuilderError"`. Raised while compiling, and sur | ------------------ | ------------------------------------------------------------------------ | | `UnresolvedParam` | A param the params bag has no value for | | `InvalidLiteral` | A param value, or a comparison operand, the column's codec rejects | -| `InvalidArguments` | Arguments a function cannot use — an empty condition list, a bad pattern | +| `InvalidArguments` | Arguments a function cannot use — an empty condition list, a bad pattern, an insert with no rows or an unknown column, more bound values than the dialect allows | ### `QueryBuilderDefect` diff --git a/docs/tables-and-types.md b/docs/tables-and-types.md index affeec5..3371cc5 100644 --- a/docs/tables-and-types.md +++ b/docs/tables-and-types.md @@ -22,6 +22,10 @@ A table is a plain value — `{ _tag: "Table", name, columns }`. It is never che live server, so a column that does not exist in ClickHouse will typecheck happily and fail at query time. Treat the declaration as a contract you keep in sync with your migrations. +The third argument takes options. `tenantColumn` names the column that carries tenancy (see +[Tenant scoping](./tenant-scoping.md)); `defaults` lists the columns the database fills in when +an insert leaves them out (see [Inserting rows](./inserts.md#which-columns-have-defaults)). + ## Column types A column type is an Effect `Schema` plus the ClickHouse type name it stands for. That schema is diff --git a/src/ch/compile.ts b/src/ch/compile.ts index 4eb2909..2217c1f 100644 --- a/src/ch/compile.ts +++ b/src/ch/compile.ts @@ -7,15 +7,16 @@ // 3. Evaluating the whereFn (with params resolved) to get Conditions // 4. Assembling into SqlQuery and calling the existing compileQuery() -import { dateTime, dateTime64, type CHType, type ColumnDefs } from "./types" +import { custom, dateTime, dateTime64, type CHType, type ColumnDefs } from "./types" import type { CHQuery, CHQueryState } from "./query" import type { CHUnionQuery } from "./union" +import { isInsert, type CHInsert } from "./insert" import { createQualifiedColumnAccessor, createJoinedColumnAccessor, sourceAlias } from "./query" -import { aliased, columnTypeOf } from "./expr" +import { aliased, columnTypeOf, isExprLike } from "./expr" import { raw, identPath, quoteIdent, quoteIdentPath, compile as compileSqlFragment } from "../sql/sql-fragment" import { splitTerminalClauses } from "../sql/terminal-clauses" import { compileQuery, type SqlQuery } from "../sql/sql-query" -import { PARAM_MARKER_PREFIX, PARAM_PLACEHOLDER_PATTERN, paramSchema, type ParamKind } from "./param" +import { PARAM_MARKER_PREFIX, PARAM_PLACEHOLDER_PATTERN, param, paramSchema, type ParamKind } from "./param" import { mergeResultSchemas } from "./define-fn" import { encodeValue } from "./literal" import { checkedLiteral, clickhouseDialect, currentDialect, withDialect, type Dialect } from "./dialect" @@ -127,6 +128,12 @@ interface ResolvedCte { interface CompiledQueryBase { readonly sql: string + /** + * What the statement does. An `insert` without RETURNING sends back no rows, + * so an executor runs it the way it runs DDL (`Database.run` does), not + * through a client path that expects a result set. + */ + readonly kind: "select" | "insert" /** * The values a binding dialect sends beside `sql`, in placeholder order. * @@ -335,6 +342,7 @@ const makeCompiledQuery = ( rawSql?: { readonly reason: string; readonly justification: string }, rowSchemaMismatch?: RowSchemaMismatch, dialect?: string, + kind: "select" | "insert" = "select", ): CompiledQuery => { let cachedDecodeRow: ((row: unknown) => Effect.Effect) | undefined let decoderBuilt = false @@ -396,6 +404,7 @@ const makeCompiledQuery = ( return { sql, + kind, parameters, tenantScope, // Resolved eagerly only here, where the getter is already memoised by @@ -457,6 +466,8 @@ export const rawCompiledQuery = < readonly route?: Route /** The `name` of the dialect the SQL is written for, so an executor can check it. */ readonly dialect?: string + /** `insert` for a write that returns no rows. Default `select`. */ + readonly kind?: "select" | "insert" }): CompiledQuery => makeCompiledQuery( args.sql, @@ -469,6 +480,7 @@ export const rawCompiledQuery = < { reason: args.reason, justification: args.justification }, undefined, args.dialect, + args.kind, ) /** @@ -505,7 +517,7 @@ const asEffect = (compile: () => A): Effect.Effect => * rather than a typed 400. Use {@link compileCHUnsafe} where a throw is what you * want — a fixture that fails to compile should fail its test loudly. */ -export const compileCH = < +export function compileCH< Cols extends ColumnDefs, Output extends Record, Joins extends Record, @@ -521,8 +533,25 @@ export const compileCH = < deferParams?: boolean dialect?: Dialect }, -): Effect.Effect, QueryBuilderError> => - asEffect(() => compileCHUnsafe(query, params, options)) +): Effect.Effect, QueryBuilderError> +/** An INSERT. `params` fills the `param.*` markers among its values. */ +export function compileCH( + insert: CHInsert, + params?: Record, + options?: InsertCompileOptions, +): Effect.Effect, QueryBuilderError> +export function compileCH( + query: CHQuery | CHInsert, + params?: Record, + options?: any, +): Effect.Effect, QueryBuilderError> { + return asEffect(() => compileCHUnsafe(query as CHQuery, params ?? {}, options)) +} + +/** What compiling an INSERT takes: only the dialect. */ +export interface InsertCompileOptions { + readonly dialect?: Dialect +} /** {@link compileCH} for a `UNION ALL`. */ export const compileUnion = , Params extends Record>( @@ -552,8 +581,21 @@ export function compileCHUnsafe< /** How params reach the server. ClickHouse literals when omitted. */ dialect?: Dialect }, -): CompiledQuery { - return withDialect(options?.dialect ?? currentDialect(), () => compileInner(query, params, options)) +): CompiledQuery +/** An INSERT. `params` fills the `param.*` markers among its values. */ +export function compileCHUnsafe( + insert: CHInsert, + params?: Record, + options?: InsertCompileOptions, +): CompiledQuery +export function compileCHUnsafe( + query: CHQuery | CHInsert, + params?: Record, + options?: any, +): CompiledQuery { + return withDialect(options?.dialect ?? currentDialect(), () => + isInsert(query) ? compileInsert(query, params ?? {}) : compileInner(query, params ?? {}, options), + ) } /** @@ -1207,6 +1249,13 @@ function renderParams( return marker }) + if (style._tag === "bind" && style.maxParameters !== undefined && parameters.length > style.maxParameters) { + throw new QueryBuilderError({ + code: "InvalidArguments", + message: `compile: the statement binds ${parameters.length} values, over the ${style.maxParameters} ${dialect.name} allows in one statement; send fewer rows per statement`, + }) + } + if (missing.length > 0) { throw new QueryBuilderError({ code: "UnresolvedParam", @@ -1260,3 +1309,122 @@ function encodeParam(dialect: Dialect, kind: ParamKind, name: string, value: unk } return encodeValue(schema, value, paramContext(kind, name)) } + +// INSERT + +/** + * The column type an insert's literal values are bound as. The value is + * encoded through its own column's codec first (so a failure names the row and + * column), and this passes the wire value on unchanged: ClickHouse writes it as + * a literal, Postgres binds it. A Postgres placeholder in `VALUES` needs no + * cast: the server coerces it to the target column. + */ +const insertWireValue = custom("insert value", Schema.Unknown) + +/** Prefix of the params an insert's literal values become. */ +const VALUE_PARAM = "$$v" + +const EMPTY_ROW = Schema.Struct({}) as unknown as CompiledQueryRowSchema + +/** + * An INSERT ... VALUES. Columns are written in table order, so two rows with + * their keys in different orders cannot swap values; a column some rows leave + * out is `DEFAULT` in those rows, which both dialects accept. + */ +function compileInsert(insert: CHInsert, params: Record): CompiledQuery { + const { table, rows } = insert._state + const where = `insertInto(${table.name})` + if (rows === undefined) throw new QueryBuilderDefect({ message: `${where}: values() is required` }) + // The rows usually come from data, so their number and keys are failures, not defects. + if (rows.length === 0) { + throw new QueryBuilderError({ code: "InvalidArguments", message: `${where}: values() was given no rows` }) + } + const computed = new Set(table.computed ?? []) + const present = new Set() + rows.forEach((row, index) => { + for (const [column, value] of Object.entries(row)) { + if (value === undefined) continue + if (!Object.hasOwn(table.columns, column)) { + throw new QueryBuilderError({ + code: "InvalidArguments", + message: `${where}: row ${index} has ${JSON.stringify(column)}, which is not a column of the table`, + }) + } + if (computed.has(column)) { + throw new QueryBuilderError({ + code: "InvalidArguments", + message: `${where}: row ${index} writes ${column}, which the database computes (MATERIALIZED or ALIAS)`, + }) + } + present.add(column) + } + }) + const columns = Object.keys(table.columns).filter((column) => present.has(column)) + if (columns.length === 0) { + throw new QueryBuilderError({ code: "InvalidArguments", message: `${where}: every row is empty; give at least one column` }) + } + + const values: Record = { ...params } + let next = 0 + const cell = (column: string, value: unknown, index: number): string => { + if (value === undefined) return "DEFAULT" + if (isExprLike(value)) return compileSqlFragment(value.toFragment()) + const wire = encodeValue(table.columns[column]!.literalSchema, value, `${where}: row ${index}, column ${column}`) + let name = `${VALUE_PARAM}${next++}` + while (Object.hasOwn(params, name)) name = `${VALUE_PARAM}${next++}` + values[name] = wire + return compileSqlFragment(param.of(insertWireValue, name).toFragment()) + } + + // A tenant table's insert is single-tenant when every row pins the tenant + // column to the same value or param. Any other expression, a NULL or a + // default could be anything. + const tenant = table.tenantColumn + const bounds = new Set() + let pinned = tenant !== undefined + + const tuples = withSubqueryCompiler( + (subquery) => + typeof subquery === "string" ? subquery : compileInner(subquery, values, { skipFormat: true, nested: true }).sql, + () => + rows.map((row, index) => { + const cells = columns.map((column) => { + const value = row[column] + const sql = cell(column, value, index) + if (column === tenant && pinned) { + if (value === undefined || value === null || (isExprLike(value) && !("_paramName" in value))) pinned = false + else bounds.add(inlineParams(sql, values)) + } + return sql + }) + if (tenant !== undefined && !present.has(tenant)) pinned = false + return `(${cells.join(", ")})` + }), + ) + + const dialect = currentDialect() + const rendered = renderParams( + `INSERT INTO ${quoteIdentPath(table.name)} (${columns.map(quoteIdent).join(", ")})\nVALUES ${tuples.join(", ")}`, + values, + dialect, + ) + const tenantScope: TenantScope = + tenant === undefined ? "untenanted" : pinned && bounds.size === 1 ? "single-tenant" : "cross-tenant" + + return withTenantBound( + makeCompiledQuery( + rendered.sql, + rendered.parameters, + tenantScope, + "derived", + () => EMPTY_ROW, + undefined, + [], + undefined, + undefined, + dialect.name, + "insert", + ), + tenantScope === "single-tenant" ? [...bounds][0] : undefined, + ) +} diff --git a/src/ch/dialect.ts b/src/ch/dialect.ts index 60160f8..f84760b 100644 --- a/src/ch/dialect.ts +++ b/src/ch/dialect.ts @@ -38,6 +38,12 @@ export type ParamStyle = * a param used twice is bound twice. */ readonly reuse: boolean + /** + * The most values one statement may bind, when the server has a limit + * (Postgres: 65535). A statement over it fails to compile, rather than + * at the server or by being split into several statements. + */ + readonly maxParameters?: number } /** Clauses that exist in some dialects and not others. */ diff --git a/src/ch/expr.ts b/src/ch/expr.ts index 3f8e658..b41c1b3 100644 --- a/src/ch/expr.ts +++ b/src/ch/expr.ts @@ -143,7 +143,7 @@ export interface Condition { // Core helpers (exported for define-fn.ts and consumer extensibility) /** An already-built expression or condition, rather than a value to encode. */ -const isExprLike = (value: unknown): value is Expr => +export const isExprLike = (value: unknown): value is Expr => value != null && typeof value === "object" && "_brand" in value && diff --git a/src/ch/index.ts b/src/ch/index.ts index c669cb6..0bca824 100644 --- a/src/ch/index.ts +++ b/src/ch/index.ts @@ -239,6 +239,9 @@ export { fromUnion, } from "./query" +// Insert builder +export { type CHInsert, type InsertRow, type InsertRowOf, type InsertValue, insertInto } from "./insert" + // Compilation export { // `compileCH` / `compileCHUnsafe` are the internal names; the public API is @@ -251,6 +254,7 @@ export { type CompiledQuery, type CompiledQueryInput, type CompiledQueryRowSchema, + type InsertCompileOptions, type RowSchemaMismatch, type TenantScope, CompiledQueryDecodeError, diff --git a/src/ch/insert.test-d.ts b/src/ch/insert.test-d.ts new file mode 100644 index 0000000..eae78e9 --- /dev/null +++ b/src/ch/insert.test-d.ts @@ -0,0 +1,92 @@ +// Type-level tests: the insert row type + +import { Schema, type DateTime } from "effect" +import { expectTypeOf } from "expect-type" +import * as CH from "./index" +import * as PG from "../postgres" +import * as S from "../schema" +import type { RowOf } from "../database" + +const Events = CH.table( + "events", + { OrgId: CH.string, Id: CH.uint64, At: CH.dateTime, Note: CH.nullable(CH.string) }, + { defaults: ["Id"] }, +) + +// Required: columns without a default that are not nullable. Optional: the +// declared defaults and the nullable columns. +expectTypeOf>().toEqualTypeOf<"OrgId" | "Id" | "At" | "Note">() +CH.insertInto(Events).values({ OrgId: "o", At: new Date() }) +CH.insertInto(Events).values({ OrgId: "o", At: "2026-01-01 00:00:00", Id: 1, Note: null }) +CH.insertInto(Events).values([{ OrgId: CH.param.string("org"), At: CH.param.dateTime("at") }]) + +// @ts-expect-error At is required +CH.insertInto(Events).values({ OrgId: "o" }) +// @ts-expect-error null only on a nullable column +CH.insertInto(Events).values({ OrgId: null, At: new Date() }) +// @ts-expect-error a value of another type +CH.insertInto(Events).values({ OrgId: "o", At: new Date(), Id: "1" }) +// @ts-expect-error not a column +CH.insertInto(Events).values({ OrgId: "o", At: new Date(), Nope: 1 }) +// @ts-expect-error a param of another type +CH.insertInto(Events).values({ OrgId: CH.param.int("org"), At: new Date() }) + +// A nullable column takes a param of its non-null type. +CH.insertInto(Events).values({ OrgId: "o", At: new Date(), Note: CH.param.string("note") }) + +// A table without `defaults` makes every non-nullable column required. +const Plain = CH.table("plain", { A: CH.string, B: CH.uint32 }) +// @ts-expect-error B is required +CH.insertInto(Plain).values({ A: "a" }) + +// A branded column takes its branded value and a plain-string param. +const OrgId = Schema.String.pipe(Schema.brand("OrgId")) +const Branded = CH.table("branded", { OrgId: CH.custom("String", OrgId) }) +CH.insertInto(Branded).values({ OrgId: OrgId.make("o") }) +CH.insertInto(Branded).values({ OrgId: CH.param.string("org") }) + +// defineTable: defaults are optional, computed columns are not in the row. +const Spans = S.defineTable("spans", { + columns: { + OrgId: CH.string, + Duration: S.column(CH.uint64, { default: 0 }), + Started: S.column(CH.dateTime, { defaultExpr: ($) => CH.rawExpr("now()", CH.dateTime) }), + Day: S.column(CH.string, { materialized: "toString(toDate(Started))" }), + Label: S.column(CH.string, { comment: "shown" }), + }, + engine: S.engine.mergeTree(), + orderBy: ["OrgId"], +}) +expectTypeOf>().toEqualTypeOf<{ + readonly OrgId: CH.InsertValue + readonly Label: CH.InsertValue + readonly Duration?: CH.InsertValue | undefined + readonly Started?: CH.InsertValue | undefined +}>() +// @ts-expect-error Day is MATERIALIZED +CH.insertInto(Spans).values({ OrgId: "o", Label: "l", Day: "x" }) +// A table with defaults still goes everywhere a table does. +CH.from(Spans).select("OrgId", "Day") +expectTypeOf(S.column(CH.string)).toEqualTypeOf>() + +// A table typed without insert metadata reads as "no defaults, nothing computed". +declare const Loose: CH.Table<"loose", { A: CH.CHString; B: CH.CHNullable }> +expectTypeOf>().toEqualTypeOf<{ + readonly A: CH.InsertValue + readonly B?: CH.InsertValue> | undefined +}>() + +// Postgres columns. +const Keys = CH.table("api_keys", { id: PG.uuid, created_at: PG.timestamptz }, { defaults: ["created_at"] }) +CH.insertInto(Keys).values({ id: "k" }) +CH.insertInto(Keys).values({ id: "k", created_at: new Date() as unknown as DateTime.Utc }) + +// An insert without RETURNING runs to no rows; compile gives a CompiledQuery. +const insert = CH.insertInto(Plain).values({ A: "a", B: 1 }) +expectTypeOf>().toEqualTypeOf() +expectTypeOf(CH.compileUnsafe(insert)).toEqualTypeOf>() +expectTypeOf(PG.compileUnsafe(insert, {})).toEqualTypeOf>() +// Queries still resolve to the query overload. +expectTypeOf(CH.compileUnsafe(CH.from(Plain).select("A"), {})).toEqualTypeOf< + CH.CompiledQuery<{ readonly A: string }, undefined> +>() diff --git a/src/ch/insert.test.ts b/src/ch/insert.test.ts new file mode 100644 index 0000000..cc4b014 --- /dev/null +++ b/src/ch/insert.test.ts @@ -0,0 +1,189 @@ +import { describe, expect, it } from "@effect/vitest" +import { DateTime, Effect, Exit } from "effect" +import * as CH from "./index" +import * as PG from "../postgres" +import * as S from "../schema" +import { QueryBuilderDefect, QueryBuilderError } from "./errors" + +const Events = CH.table( + "events", + { + OrgId: CH.string, + Id: CH.uint64, + At: CH.dateTime, + Attrs: CH.map(CH.string, CH.string), + Note: CH.nullable(CH.string), + Tags: CH.array(CH.string), + }, + { tenantColumn: "OrgId", defaults: ["Id"] }, +) + +const Keys = CH.table( + "api_keys", + { id: PG.uuid, org_id: PG.text, name: PG.text, created_at: PG.timestamptz, revoked: PG.bool, meta: PG.jsonb() }, + { defaults: ["created_at", "revoked"] }, +) + +const failure = (exit: Exit.Exit) => + Exit.isFailure(exit) ? exit.cause.reasons.map((r) => ("error" in r ? r.error : "defect" in r ? r.defect : r))[0] : undefined + +describe("insertInto", () => { + it("writes columns in table order whatever the key order, and DEFAULT for a missing key", () => { + const compiled = CH.compileUnsafe( + CH.insertInto(Events).values([ + { Tags: ["x"], Note: null, Attrs: { a: "it's" }, At: new Date(0), OrgId: "o1" }, + { OrgId: "o1", Id: 5, At: "2026-01-01 00:00:00", Attrs: {}, Note: "n", Tags: [] }, + ]), + ) + expect(compiled.sql).toBe( + "INSERT INTO events (OrgId, Id, At, Attrs, Note, Tags)\n" + + "VALUES ('o1', DEFAULT, '1970-01-01 00:00:00', map('a', 'it\\'s'), NULL, ['x']), " + + "('o1', 5, '2026-01-01 00:00:00', map(), 'n', [])", + ) + expect(compiled.kind).toBe("insert") + expect(compiled.parameters).toEqual([]) + expect(compiled.dialect).toBe("clickhouse") + }) + + it("leaves out a column no row gives", () => { + const compiled = CH.compileUnsafe( + CH.insertInto(Events).values({ OrgId: "o", At: DateTime.makeUnsafe(0), Attrs: {}, Tags: [] }), + ) + expect(compiled.sql).toBe("INSERT INTO events (OrgId, At, Attrs, Tags)\nVALUES ('o', '1970-01-01 00:00:00', map(), [])") + }) + + it("binds every value on Postgres, one placeholder per param", () => { + const compiled = PG.compileUnsafe( + CH.insertInto(Keys).values([ + { id: CH.param.string("id"), org_id: CH.param.string("org"), name: "a", meta: { k: 1 } }, + { id: "b", org_id: CH.param.string("org"), name: "it's", revoked: true, meta: null }, + ]), + { id: "a", org: "o1" }, + ) + expect(compiled.sql).toBe( + 'INSERT INTO "api_keys" ("id", "org_id", "name", "revoked", "meta")\n' + + "VALUES ($1, $2, $3, DEFAULT, $4), ($5, $2, $6, $7, $8)", + ) + expect(compiled.parameters).toEqual(["a", "o1", "a", '{"k":1}', "b", "it's", true, "null"]) + expect(compiled.dialect).toBe("postgres") + }) + + it("writes an expression as SQL", () => { + const Stamps = CH.table("stamps", { At: CH.dateTime, N: CH.uint64 }) + const compiled = CH.compileUnsafe( + CH.insertInto(Stamps).values({ At: CH.rawExpr("now()", CH.dateTime), N: CH.toUInt64(CH.lit("7")) }), + ) + expect(compiled.sql).toBe("INSERT INTO stamps (At, N)\nVALUES (now(), toUInt64('7'))") + }) + + it("is immutable: values replaces, and a later push to the array changes nothing", () => { + const rows = [{ OrgId: "a", At: new Date(0), Attrs: {}, Tags: [] }] + const base = CH.insertInto(Events) + const first = base.values(rows) + rows.push({ OrgId: "b", At: new Date(0), Attrs: {}, Tags: [] }) + const second = first.values({ OrgId: "c", At: new Date(0), Attrs: {}, Tags: [] }) + expect(CH.compileUnsafe(first).sql).toContain("('a'") + expect(CH.compileUnsafe(first).sql).not.toContain("'b'") + expect(CH.compileUnsafe(second).sql).toContain("('c'") + expect(CH.compileUnsafe(second).sql).not.toContain("'a'") + expect(base._state.rows).toBeUndefined() + }) + + it("a value that could spell a param marker stays a value", () => { + const Notes = CH.table("notes", { Body: CH.string }) + const compiled = CH.compileUnsafe(CH.insertInto(Notes).values({ Body: "__PARAM_string_x__" }), { x: "boom" }) + expect(compiled.sql).toBe("INSERT INTO notes (Body)\nVALUES ('\\x5F_PARAM_string_x__')") + }) + + describe("tenant scope", () => { + const scope = (rows: ReadonlyArray>, params: Record = {}) => + CH.compileUnsafe(CH.insertInto(Events).values(rows as any), params).tenantScope + const row = { At: new Date(0), Attrs: {}, Tags: [] } + + it("single-tenant when every row gives the same value or param", () => { + expect(scope([{ ...row, OrgId: "o" }, { ...row, OrgId: "o" }])).toBe("single-tenant") + expect(scope([{ ...row, OrgId: CH.param.string("org") }], { org: "o" })).toBe("single-tenant") + // The same tenant as a literal and as a param. + expect(scope([{ ...row, OrgId: CH.param.string("org") }, { ...row, OrgId: "o" }], { org: "o" })).toBe( + "single-tenant", + ) + }) + + it("cross-tenant when rows differ or a row cannot be read", () => { + expect(scope([{ ...row, OrgId: "a" }, { ...row, OrgId: "b" }])).toBe("cross-tenant") + expect(scope([{ ...row, OrgId: CH.rawExpr("'a'", CH.string) }])).toBe("cross-tenant") + }) + + it("untenanted for a table without a tenant column", () => { + const Plain = CH.table("plain", { A: CH.string }) + expect(CH.compileUnsafe(CH.insertInto(Plain).values({ A: "x" })).tenantScope).toBe("untenanted") + }) + }) + + describe("failures", () => { + it.effect("no rows, an unknown column, an empty row, and a bad value fail in the error channel", () => + Effect.gen(function* () { + const Plain = CH.table("plain", { A: CH.uint32, B: CH.nullable(CH.string) }) + const errors = yield* Effect.all( + [ + CH.compile(CH.insertInto(Plain).values([])), + CH.compile(CH.insertInto(Plain).values({ A: 1, C: 2 } as any)), + CH.compile(CH.insertInto(Plain).values([{ B: undefined } as any])), + CH.compile(CH.insertInto(Plain).values({ A: null as any })), + CH.compile(CH.insertInto(Plain).values({ A: CH.param.int("a") })), + ].map(Effect.flip), + ) + for (const error of errors) expect(error).toBeInstanceOf(QueryBuilderError) + expect(errors.map((e) => e.code)).toEqual([ + "InvalidArguments", + "InvalidArguments", + "InvalidArguments", + "InvalidLiteral", + "UnresolvedParam", + ]) + expect(errors[1]!.message).toContain('row 0 has "C"') + expect(errors[3]!.message).toContain("row 0, column A") + }), + ) + + it.effect("compiling without values is a defect", () => + Effect.gen(function* () { + const exit = yield* Effect.exit(CH.compile(CH.insertInto(Events))) + expect(failure(exit)).toBeInstanceOf(QueryBuilderDefect) + }), + ) + + it("a statement over Postgres's parameter limit fails to compile", () => { + const Wide = CH.table("wide", { A: PG.int4 }) + const rows = Array.from({ length: 65536 }, (_, i) => ({ A: i })) + expect(() => PG.compileUnsafe(CH.insertInto(Wide).values(rows))).toThrow(/65536 values, over the 65535/) + expect(PG.compileUnsafe(CH.insertInto(Wide).values(rows.slice(1))).parameters).toHaveLength(65535) + }) + }) + + describe("defineTable", () => { + const Spans = S.defineTable("spans", { + columns: { + OrgId: CH.string, + Duration: S.column(CH.uint64, { default: 0 }), + Started: S.column(CH.dateTime, { defaultExpr: "now()" }), + Day: S.column(CH.string, { materialized: "toString(toDate(Started))" }), + Label: S.column(CH.string, { comment: "shown in the UI" }), + }, + engine: S.engine.mergeTree(), + orderBy: ["OrgId"], + }) + + it("records which columns have defaults and which are computed", () => { + expect(Spans.defaults).toEqual(["Duration", "Started"]) + expect(Spans.computed).toEqual(["Day"]) + }) + + it.effect("refuses a computed column at runtime too", () => + Effect.gen(function* () { + const error = yield* Effect.flip(CH.compile(CH.insertInto(Spans).values({ OrgId: "o", Label: "l", Day: "x" } as any))) + expect(error.message).toContain("writes Day, which the database computes") + }), + ) + }) +}) diff --git a/src/ch/insert.ts b/src/ch/insert.ts new file mode 100644 index 0000000..adf7b0f --- /dev/null +++ b/src/ch/insert.ts @@ -0,0 +1,117 @@ +// Insert Builder +// +// `insertInto(table).values(rows)` describes an INSERT the way `from(table)` +// describes a SELECT: an immutable value read by `compile`, which writes it for +// the dialect it is given. Values are encoded through each column's own codec, +// the same one that decodes it. See `design/writes.md`. +// +// Usage: +// const insert = CH.insertInto(ApiKeys).values({ +// id: CH.param.string("id"), +// orgId: CH.param.string("orgId"), +// name: "default", +// }) +// yield* Database.run(insert, { id, orgId }) + +import type { Comparable, Expr, Widen } from "./expr" +import type { Table } from "./table" +import type { CHType, ColumnDefs, InferTS } from "./types" + +/** + * What an insert may write into a column: a value of the column's type, a + * `param.*` of it, or any expression of it (`CH.now()`, `rawExpr(...)`). + * + * A value is the decoded type (a branded id stays branded), plus the extra + * forms a comparison accepts (`Date` or a string for a `DateTime`). A param or + * expression may be of the widened primitive, as in a comparison, so a branded + * column takes `param.string`. + */ +export type InsertValue> = + | Comparable> + | Expr> + | Expr>> + | Expr>>> + +/** A bare `string` means the table did not say, which is read as none. */ +type Known = string extends K ? never : K + +type NullableKeys = { + [K in keyof Cols]: null extends InferTS ? K : never +}[keyof Cols] & + string + +type OptionalKeys = Known | NullableKeys + +type Simplify = { [K in keyof A]: A[K] } & {} + +/** + * One row of an insert into a table with these columns. + * + * A column is optional when it is nullable or the table declares a default for + * it; leaving it out (or passing `undefined`) writes the column's default. + * `null` writes NULL and only type-checks on a nullable column. Computed + * columns (ClickHouse `MATERIALIZED` and `ALIAS`) are not in the row at all. + */ +export type InsertRow = Simplify< + { + readonly [K in Exclude | Known>]: InsertValue + } & { + readonly [K in Exclude & keyof Cols, Known>]?: InsertValue | undefined + } +> + +/** The insert row of a table value: `InsertRowOf`. */ +export type InsertRowOf = + T extends Table ? InsertRow : never + +/** @internal — runtime insert state */ +export interface CHInsertState { + readonly table: Table + /** Set by `values`. Compiling without it is a defect. */ + readonly rows?: ReadonlyArray>> +} + +export interface CHInsert< + Cols extends ColumnDefs = ColumnDefs, + Defaulted extends string = never, + Computed extends string = never, + Output = never, +> { + readonly _tag: "CHInsert" + /** @internal — runtime insert state */ + readonly _state: CHInsertState + /** phantom. `output` is the row `Database.run` returns: none without RETURNING. */ + readonly _phantom?: { readonly cols: Cols; readonly output: Output } + + /** + * The rows to insert: one row or an array. Calling it again replaces the + * rows. Columns are written in table order whatever the key order, and a + * column some rows leave out is written as `DEFAULT` in those rows. + */ + values( + rows: InsertRow | ReadonlyArray>, + ): CHInsert +} + +const makeInsert = ( + state: CHInsertState, +): CHInsert => ({ + _tag: "CHInsert", + _state: state, + values: (rows) => + makeInsert({ + ...state, + // Copied, so a caller pushing to its array later does not change the insert. + rows: Array.isArray(rows) ? [...rows] : [rows as Readonly>], + }), +}) + +/** Start an INSERT into `table`. Give its rows with `values`. */ +export function insertInto( + table: Table, +): CHInsert { + return makeInsert({ table: table as Table }) +} + +export const isInsert = (value: unknown): value is CHInsert => + typeof value === "object" && value !== null && (value as { readonly _tag?: unknown })._tag === "CHInsert" diff --git a/src/ch/table.ts b/src/ch/table.ts index 558c846..853bd86 100644 --- a/src/ch/table.ts +++ b/src/ch/table.ts @@ -5,7 +5,19 @@ import type { ColumnDefs } from "./types" -export interface Table { +/** + * `Defaulted` and `Computed` describe inserts: the columns an insert may leave + * out because the database fills them, and the ones it may not write at all + * (ClickHouse `MATERIALIZED` and `ALIAS`). Both default to `string`, read as + * "unknown", so a `Table` with more of either still satisfies a `Table` + * written without them; the insert row type treats a bare `string` as none. + */ +export interface Table< + Name extends string, + Columns extends ColumnDefs, + Defaulted extends string = string, + Computed extends string = string, +> { readonly _tag: "Table" readonly name: Name readonly columns: Columns @@ -21,21 +33,33 @@ export interface Table { * satisfy a `Table` type declared with fewer. */ readonly tenantColumn?: string + /** Columns with a database default, which an insert may leave out. */ + readonly defaults?: ReadonlyArray + /** Columns the database computes, which an insert may not write. */ + readonly computed?: ReadonlyArray } -export interface TableOptions { +export interface TableOptions { readonly tenantColumn?: keyof Columns & string + /** + * Columns the database fills when an insert leaves them out: a Postgres + * `serial` or `DEFAULT now()`, a ClickHouse `DEFAULT`. Nullable columns are + * optional in an insert without being listed. `defineTable` works this out + * from its column options. + */ + readonly defaults?: ReadonlyArray } -export function table( - name: Name, - columns: Columns, - options?: TableOptions, -): Table { +export function table< + const Name extends string, + const Columns extends ColumnDefs, + const Defaulted extends keyof Columns & string = never, +>(name: Name, columns: Columns, options?: TableOptions): Table { return { _tag: "Table", name, columns, ...(options?.tenantColumn !== undefined ? { tenantColumn: options.tenantColumn } : undefined), + ...(options?.defaults !== undefined && options.defaults.length > 0 ? { defaults: [...options.defaults] } : undefined), } } diff --git a/src/database/database.test.ts b/src/database/database.test.ts index b3d8d3b..9cabef3 100644 --- a/src/database/database.test.ts +++ b/src/database/database.test.ts @@ -5,7 +5,7 @@ import { PgliteClient } from "@effect/sql-pglite" import { assert, describe, expect, it, layer } from "@effect/vitest" -import { Cause, Deferred, Effect, Exit, Fiber, Layer, Ref, Schedule, Schema } from "effect" +import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Ref, Schedule, Schema } from "effect" import * as SqlClient from "effect/sql/SqlClient" import * as CH from "../index" import * as PG from "../postgres" @@ -366,6 +366,65 @@ layer(Live, { excludeTestServices: true })("Database on PGlite", (it) => { expect(yield* ids(table)).toEqual([1]) }), ) + + it.effect("run inserts rows built with insertInto, bound and decoded back through the column codecs", () => + Effect.gen(function* () { + yield* Db.execute( + Db.sql`CREATE TABLE keys ( + id uuid PRIMARY KEY, + org_id text NOT NULL, + uses int8 NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now(), + revoked boolean NOT NULL DEFAULT false, + meta jsonb, + tags text[] NOT NULL, + note text + )`, + ) + const Keys = CH.table( + "keys", + { + id: PG.uuid, + org_id: PG.text, + uses: PG.int8, + created_at: PG.timestamptz, + revoked: PG.bool, + meta: PG.nullable(PG.jsonb()), + tags: PG.array(PG.text), + note: PG.nullable(PG.text), + }, + { tenantColumn: "org_id", defaults: ["uses", "created_at", "revoked"] }, + ) + const at = DateTime.makeUnsafe("2026-01-02T03:04:05.678Z") + const id = (n: number) => `00000000-0000-0000-0000-00000000000${n}` + const inserted = yield* Db.run( + CH.insertInto(Keys).values([ + { id: id(1), org_id: CH.param.string("org"), tags: ["a", "it's"], meta: { k: [1, 2] }, created_at: at }, + { id: id(2), org_id: CH.param.string("org"), tags: [], uses: 7, revoked: true, note: "n" }, + ]), + { org: "o1" }, + ) + expect(inserted).toEqual([]) + const rows = yield* Db.run(CH.from(Keys).select("id", "uses", "created_at", "revoked", "meta", "tags", "note").orderBy(["id", "asc"])) + expect(rows[0]).toEqual({ id: id(1), uses: 0, created_at: at, revoked: false, meta: { k: [1, 2] }, tags: ["a", "it's"], note: null }) + expect(rows[1]).toMatchObject({ id: id(2), uses: 7, revoked: true, meta: null, tags: [], note: "n" }) + }), + ) + + it.effect("an insert inside a failed transaction rolls back", () => + Effect.gen(function* () { + const table = yield* freshTable + const T = CH.table(table, { id: PG.int4, note: PG.nullable(PG.text) }) + const exit = yield* Effect.exit( + Db.transaction( + Effect.andThen(Db.run(CH.insertInto(T).values({ id: 1 })), Effect.fail(new Domain())), + ), + ) + expect(Exit.isFailure(exit)).toBe(true) + yield* Db.run(CH.insertInto(T).values([{ id: 2 }, { id: 3, note: "x" }])) + expect(yield* ids(table)).toEqual([2, 3]) + }), + ) }) describe("sql templates per dialect", () => { diff --git a/src/database/database.ts b/src/database/database.ts index a6a78f8..6d31464 100644 --- a/src/database/database.ts +++ b/src/database/database.ts @@ -14,6 +14,7 @@ import { isSqlError } from "effect/sql/SqlError" import { compileCH, compileUnion, CompiledQueryDecodeError, type CompiledQuery } from "../ch/compile" import { noTransactions, type Dialect, type IsolationLevel, type TransactionSettings } from "../ch/dialect" import type { QueryBuilderError } from "../ch/errors" +import type { CHInsert } from "../ch/insert" import type { CHQuery } from "../ch/query" import type { CHUnionQuery } from "../ch/union" import { @@ -37,8 +38,8 @@ export interface Statement { /** What `query` and `execute` take: a `sql\`...\`` template, or SQL with its parameters. */ export type StatementInput = SqlTemplate | Statement -/** What `run` takes: a built query, a union, or a query already compiled. */ -export type Runnable = CHQuery | CHUnionQuery | CompiledQuery +/** What `run` takes: a built query, a union, an insert, or one already compiled. */ +export type Runnable = CHQuery | CHUnionQuery | CHInsert | CompiledQuery /** The decoded row of a `Runnable`. */ export type RowOf = @@ -84,6 +85,7 @@ export interface DatabaseApi { readonly dialect: Dialect /** * Compile a query for this database's dialect, run it, and decode its rows. + * An insert returns no rows. * `params` fills the query's `param.*` markers. A query compiled elsewhere * runs as it is, if it was compiled for this dialect. */ @@ -256,14 +258,18 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp ) : Effect.succeed(runnable) } - return "_tag" in runnable && runnable._tag === "CHUnionQuery" - ? compileUnion(runnable, params, { dialect }) - : compileCH(runnable as CHQuery, params, { dialect }) + if ("_tag" in runnable && runnable._tag === "CHUnionQuery") return compileUnion(runnable, params, { dialect }) + if ("_tag" in runnable && runnable._tag === "CHInsert") return compileCH(runnable, params, { dialect }) + return compileCH(runnable as CHQuery, params, { dialect }) } + // An insert sends back no rows, so it runs the way `execute` does: through + // `command`, which a ClickHouse client needs for a statement with no result set. const run: DatabaseApi["run"] = (runnable, params = {}) => Effect.flatMap(compileFor(runnable, params), (compiled) => - Effect.flatMap(rows(compiled), (wire) => compiled.decodeRows(wire)), + compiled.kind === "insert" + ? Effect.as(execute(compiled), []) + : Effect.flatMap(rows(compiled), (wire) => compiled.decodeRows(wire)), ) const retryContention = (effect: Effect.Effect, retry?: RetryOptions) => diff --git a/src/pg/dialect.ts b/src/pg/dialect.ts index 896e9eb..a65cffb 100644 --- a/src/pg/dialect.ts +++ b/src/pg/dialect.ts @@ -94,6 +94,8 @@ export const postgresDialect: Dialect = { _tag: "bind", placeholder: (index, kind) => (Object.hasOwn(placeholderCasts, kind) ? `$${index}::${placeholderCasts[kind]}` : `$${index}`), reuse: true, + // The wire protocol counts parameters in an Int16. + maxParameters: 65535, }, clauses: { format: false, derivedTableAlias: true, groupByAlias: false, parenthesizeUnionBranches: true }, paramCodecs: { diff --git a/src/postgres.ts b/src/postgres.ts index 1937dd3..5bed08a 100644 --- a/src/postgres.ts +++ b/src/postgres.ts @@ -18,12 +18,14 @@ export * from "./pg/types" export * from "./pg/functions" /** `compile` from the root entry, for Postgres unless `options.dialect` says otherwise. */ -export const compile: typeof compileCH = (query, params, options) => - compileCH(query, params, { ...options, dialect: options?.dialect ?? postgresDialect }) +// Typed through `any` and cast: `compileCH` is overloaded (queries and inserts), +// and an overloaded type gives an arrow's parameters no contextual type. +export const compile = ((query: any, params?: any, options?: any) => + compileCH(query, params, { ...options, dialect: options?.dialect ?? postgresDialect })) as typeof compileCH /** `compileUnsafe` from the root entry, for Postgres. */ -export const compileUnsafe: typeof compileCHUnsafe = (query, params, options) => - compileCHUnsafe(query, params, { ...options, dialect: options?.dialect ?? postgresDialect }) +export const compileUnsafe = ((query: any, params?: any, options?: any) => + compileCHUnsafe(query, params, { ...options, dialect: options?.dialect ?? postgresDialect })) as typeof compileCHUnsafe /** `compileUnion` from the root entry, for Postgres. */ export const compileUnion: typeof compileUnionCH = (union, params, options) => diff --git a/src/schema.ts b/src/schema.ts index 1677959..3c62fc9 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -16,6 +16,8 @@ export { type ColumnOptions, type ColumnSpec, type ColumnsOf, + type ComputedColumnsOf, + type DefaultedColumnsOf, type DdlExpr, type DdlKey, type IndexSpec, diff --git a/src/schema/define.ts b/src/schema/define.ts index 205a84b..1800ffe 100644 --- a/src/schema/define.ts +++ b/src/schema/define.ts @@ -69,31 +69,61 @@ export interface ColumnOptions> { readonly comment?: string } -export interface ColumnSpec> { +/** + * `Given` keeps which options were given, so the table can tell an insert + * which columns it may leave out (a default) or may not write (computed). + */ +export interface ColumnSpec, Given extends keyof ColumnOptions = keyof ColumnOptions> { readonly _tag: "ColumnSpec" readonly type: T readonly options: ColumnOptions + readonly _given?: Given } /** A column with DDL options. A bare column type works too where no option is needed. */ -export const column = >(type: T, options: ColumnOptions = {}): ColumnSpec => ({ +export const column = , Given extends keyof ColumnOptions = never>( + type: T, + // Only the option *names* are inferred, from the mapped half; the options + // themselves stay contextually typed, so `materialized: ($) => ...` keeps `$`. + options: ColumnOptions & { readonly [K in Given]: unknown } = {} as ColumnOptions & { readonly [K in Given]: unknown }, +): ColumnSpec => ({ _tag: "ColumnSpec", type, options, }) -export type ColumnInput = CHType | ColumnSpec> +export type ColumnInput = CHType | ColumnSpec, any> /** The query-side column types of a `columns` record. */ export type ColumnsOf> = { - readonly [K in keyof I]: I[K] extends ColumnSpec + readonly [K in keyof I]: I[K] extends ColumnSpec ? T : I[K] extends CHType ? I[K] : never } -const isColumnSpec = (input: ColumnInput): input is ColumnSpec> => +/** Columns declared with `default` or `defaultExpr`: an insert may leave them out. */ +export type DefaultedColumnsOf> = { + [K in keyof I]: I[K] extends ColumnSpec + ? [Extract] extends [never] + ? never + : K + : never +}[keyof I] & + string + +/** Columns declared `materialized` or `alias`: an insert may not write them. */ +export type ComputedColumnsOf> = { + [K in keyof I]: I[K] extends ColumnSpec + ? [Extract] extends [never] + ? never + : K + : never +}[keyof I] & + string + +const isColumnSpec = (input: ColumnInput): input is ColumnSpec, any> => "_tag" in input && input._tag === "ColumnSpec" // Engines @@ -172,7 +202,12 @@ export interface TableDdl { readonly indexes: ReadonlyArray } -export interface SchemaTable extends Table { +export interface SchemaTable< + Name extends string, + Cols extends ColumnDefs, + Defaulted extends string = string, + Computed extends string = string, +> extends Table { readonly ddl: TableDdl } @@ -227,7 +262,7 @@ const columnDefault = ( export function defineTable>( name: Name, definition: TableDefinition, -): SchemaTable> { +): SchemaTable, DefaultedColumnsOf, ComputedColumnsOf> { assertIdentifier(name, name) const inputs = Object.entries(definition.columns) if (inputs.length === 0) throw new SchemaDefinitionDefect({ object: name, message: "a table needs columns" }) @@ -255,7 +290,9 @@ export function defineTable { assertIdentifier(`${name}.${column}`, column) - const spec = isColumnSpec(input) ? input : { _tag: "ColumnSpec" as const, type: input, options: {} } + const spec: ColumnSpec> = isColumnSpec(input) + ? input + : { _tag: "ColumnSpec", type: input as CHType, options: {} } return { kind: "column", table: name, @@ -299,11 +336,17 @@ export function defineTable c.default?.kind === "DEFAULT").map((c) => c.name) + const computed = columnEntities + .filter((c) => c.default?.kind === "MATERIALIZED" || c.default?.kind === "ALIAS") + .map((c) => c.name) return { _tag: "Table", name, columns: types, ...(definition.tenantColumn !== undefined ? { tenantColumn: definition.tenantColumn } : undefined), + ...(defaults.length > 0 ? { defaults: defaults as unknown as Array> } : undefined), + ...(computed.length > 0 ? { computed: computed as unknown as Array> } : undefined), ddl: { table, columns: columnEntities, indexes: indexEntities }, } } diff --git a/tests/database.clickhouse.test.ts b/tests/database.clickhouse.test.ts index 9de4d4c..1d0a2eb 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 { Effect, Exit } from "effect" +import { DateTime, Effect, Exit } from "effect" import { describe, expect, it } from "vitest" import * as CH from "@maple-dev/effect-orm" import * as Db from "@maple-dev/effect-orm/database" @@ -50,6 +50,55 @@ describe("database", () => { ]) }) + it("runs an insert built with insertInto through the command path", async () => { + const { rows, sent } = await Effect.runPromise( + withDatabase((db, sent) => + Effect.gen(function* () { + yield* db.execute( + Db.sql`CREATE TABLE events ( + OrgId String, + Id UInt64 DEFAULT 42, + At DateTime64(3), + Attrs Map(String, String), + Note Nullable(String), + Tags Array(String), + Day String MATERIALIZED toString(toDate(At)) + ) ENGINE = MergeTree ORDER BY (OrgId, Id)`, + ) + const Events = CH.table( + "events", + { + OrgId: CH.string, + Id: CH.uint64, + At: CH.dateTime64, + Attrs: CH.map(CH.string, CH.string), + Note: CH.nullable(CH.string), + Tags: CH.array(CH.string), + }, + { tenantColumn: "OrgId", defaults: ["Id"] }, + ) + const inserted = yield* db.run( + CH.insertInto(Events).values([ + { OrgId: CH.param.string("org"), At: new Date("2026-01-02T03:04:05.678Z"), Attrs: { a: "it's; x" }, Tags: ["t"], Note: null }, + { OrgId: CH.param.string("org"), Id: 7, At: "2026-01-02 00:00:00", Attrs: {}, Tags: [], Note: "n" }, + ]), + { org: "o1" }, + ) + expect(inserted).toEqual([]) + const rows = yield* db.run( + CH.from(Events).select("OrgId", "Id", "At", "Attrs", "Note", "Tags").orderBy(["Id", "asc"]), + ) + return { rows, sent: [...sent] } + }), + ), + ) + expect(rows.map((row) => ({ ...row, At: DateTime.formatIso(row.At) }))).toEqual([ + { OrgId: "o1", Id: 7, At: "2026-01-02T00:00:00.000Z", Attrs: {}, Note: "n", Tags: [] }, + { OrgId: "o1", Id: 42, At: "2026-01-02T03:04:05.678Z", Attrs: { a: "it's; x" }, Note: null, Tags: ["t"] }, + ]) + expect(sent[1]).toMatch(/^INSERT INTO events \(OrgId, Id, At, Attrs, Note, Tags\)\nVALUES \('o1', DEFAULT, /) + }) + it("refuses a transaction before sending anything", async () => { const result = await Effect.runPromise( withDatabase((db, sent) =>