Add onConflictDoNothing and onConflictDoUpdate to inserts

CHANGELOG.md

@@ -13,6 +13,9 @@
 - Add `returning` to an insert (Postgres): column names or a callback, as in `select`. `run`
   returns the inserted rows decoded through the derived row schema; `CompiledQuery.returning`
   lists the aliases. Add `DialectClauses.returning`, optional, absent meaning no.
+- Add `onConflictDoNothing` and `onConflictDoUpdate` to an insert (Postgres), with Drizzle's
+  options: `target` (columns or `{ constraint }`), `targetWhere`, `set` (a record, or a callback
+  over the existing row and `excluded`) and `where`. Add `DialectClauses.onConflict`, optional.
 - 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

design/writes.md

@@ -1,7 +1,7 @@
 # Writes: INSERT
 
-Status: phases 1 and 2 built (`insertInto(...).values(...).returning(...)`, see
-`docs/inserts.md`); phases 3 and 4 not started. Section 11 lists where the build differs from the plan. UPDATE and DELETE come later and
+Status: phases 1 to 3 built (`values`, `returning`, `onConflictDoNothing` /
+`onConflictDoUpdate`, see `docs/inserts.md`); phase 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).
 
@@ -245,3 +245,8 @@ note, reusing `kind: "write"`, the returning path and the `set` record type from
   optional so a dialect written against the earlier interface still compiles.
 - **`CompiledQuery.returning`** lists the RETURNING aliases; `run` reads it to choose between
   the command path and the row path, so a precompiled insert runs correctly too.
+- **`ON CONFLICT` takes Drizzle's shape**, `onConflictDoNothing({ target?, targetWhere? })` and
+  `onConflictDoUpdate({ target, targetWhere?, set, where? })`, instead of the chained
+  `.onConflict(target).doUpdate(...)` in §6: Maple's 68 call sites then move over with renames.
+  `$` in `set` and `where` is qualified with the table name (`"counters"."count"`), because an
+  unqualified column there is ambiguous with `excluded`.

docs/inserts.md

@@ -109,11 +109,52 @@ The row schema is derived from the list, as it is from a SELECT: an untyped expr
 `untypedColumns`. `CompiledQuery.returning` lists the aliases. ClickHouse has no RETURNING, so
 compiling an insert with `returning` for it is a `QueryBuilderDefect`.
 
+## On conflict
+
+On Postgres, `onConflictDoNothing` and `onConflictDoUpdate` add an `ON CONFLICT` clause. Their
+options follow Drizzle's, so code moving from Drizzle changes little.
+
+```ts
+const Counters = CH.table("counters", { key: PG.text, count: PG.int8, locked: PG.bool }, { defaults: ["locked"] })
+
+// Skip a row whose key exists. Without `target`, any unique index or constraint counts.
+CH.insertInto(Counters).values({ key: "a", count: 1 }).onConflictDoNothing({ target: ["key"] })
+
+// Upsert: add to the existing count, unless the row is locked.
+CH.insertInto(Counters)
+	.values({ key: "a", count: 1 })
+	.onConflictDoUpdate({
+		target: ["key"],
+		set: ($, excluded) => ({ count: $.count.add(excluded.count) }),
+		where: ($) => $.locked.eq(false),
+	})
+	.returning("key", "count")
+```
+
+```sql
+INSERT INTO "counters" ("key", "count")
+VALUES ($1, $2)
+ON CONFLICT ("key") DO UPDATE SET "count" = "counters"."count" + "excluded"."count" WHERE "counters"."locked" = FALSE
+RETURNING "key" AS "key", "count" AS "count"
+```
+
+- `target` is column names, or `{ constraint: "name" }`. `targetWhere` gives a partial unique
+  index's predicate. `onConflictDoUpdate` requires a `target`; `onConflictDoNothing` does not.
+- `set` is a record of values, params or expressions, or a callback that gets `$` (the existing
+  row) and `excluded` (the row proposed for insertion). `$` is qualified with the table name,
+  because an unqualified column would be ambiguous with `excluded`. A key left out keeps the
+  existing value.
+- `where` limits the update to existing rows it holds for. A row it skips is not updated and,
+  with `returning`, returns nothing; the same goes for a row `onConflictDoNothing` skips.
+- Calling either again replaces the clause. ClickHouse has no `ON CONFLICT` (deduplicate with a
+  `ReplacingMergeTree` instead), so compiling one for it is a `QueryBuilderDefect`.
+
 ## 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
+same param, and `"cross-tenant"` when rows differ or a row uses another expression. An
+`onConflictDoUpdate` that sets the tenant column counts as one more row. A table
 without a tenant column gives `"untenanted"`.
 
 ## Failures
@@ -127,6 +168,8 @@ without a tenant column gives `"untenanted"`.
 | Over the dialect's bound-value limit              | `QueryBuilderError` `InvalidArguments`   |
 | Compiling without `values`                        | `QueryBuilderDefect`                     |
 | `returning` for a dialect without RETURNING       | `QueryBuilderDefect`                     |
+| `onConflictDoUpdate` setting no or unknown columns | `QueryBuilderError` `InvalidArguments`  |
+| `onConflict*` without ON CONFLICT, a bad target   | `QueryBuilderDefect`                     |
 
 _(Backed by `src/ch/insert.test.ts`, `src/database/database.test.ts` and
 `tests/database.clickhouse.test.ts`.)_

docs/reference.md

@@ -62,7 +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, `.returning(...)` the RETURNING list (Postgres). See [Inserting rows](./inserts.md) |
+| `insertInto` | `(table) => CHInsert`; `.values(row \| rows)` sets its rows, `.returning(...)` the RETURNING list, `.onConflictDoNothing(options?)` / `.onConflictDoUpdate(options)` the ON CONFLICT clause (Postgres). See [Inserting rows](./inserts.md) |
 
 ### `CHQuery` methods
 
@@ -279,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`, `CHInsert`, `InsertRow`, `InsertRowOf`, `InsertValue`, `ColumnAccessor`, `JoinedColumnAccessor`,
+`ParamKind`, `CHQuery`, `CHUnionQuery`, `CHInsert`, `InsertRow`, `InsertRowOf`, `InsertValue`, `ConflictTarget`, `ConflictSet`, `OnConflictDoNothing`, `OnConflictDoUpdate`, `ColumnAccessor`, `JoinedColumnAccessor`,
 `JoinOnCallback`, `CompiledQuery`, `CompiledQueryInput`, `CompiledQueryRowSchema`, `RowSchemaMismatch`, `TenantScope`, `Dialect`, `DialectClauses`, `DialectTransactions`, `IsolationLevel`, `TransactionSettings`, `ParamStyle`, `FnResult`,
 `WindowFunnelMode`, `WindowSpec`, `WindowRowsFrame`, `WindowFrameBound`,
 `WindowOrderDirection`, `CompiledWindowSpec`.

src/ch/compile.ts

@@ -1333,6 +1333,74 @@ const VALUE_PARAM = "$$v"
 
 const EMPTY_ROW = Schema.Struct({}) as unknown as CompiledQueryRowSchema<never>
 
+/**
+ * The ON CONFLICT clause, or `""` without one. `$` in SET and WHERE is the
+ * existing row, qualified with the table name: an unqualified column there
+ * would be ambiguous with `excluded`.
+ */
+const onConflictClause = (
+	insert: CHInsert<any, any, any, any>,
+	cell: (column: string, value: unknown, context: string) => string,
+	wrote: (column: string, value: unknown, sql: string) => void,
+): string => {
+	const { table, conflict } = insert._state
+	if (conflict === undefined) return ""
+	const where = `insertInto(${table.name})`
+	const dialect = currentDialect()
+	if (dialect.clauses.onConflict !== true) {
+		throw new QueryBuilderDefect({
+			message: `${where}: onConflict has no meaning for the ${dialect.name} dialect, which has no ON CONFLICT clause`,
+		})
+	}
+
+	let target = ""
+	if (conflict.target !== undefined) {
+		if ("constraint" in conflict.target) {
+			if (conflict.targetWhere !== undefined) {
+				throw new QueryBuilderDefect({ message: `${where}: targetWhere needs a column target, not a constraint` })
+			}
+			target = ` ON CONSTRAINT ${quoteIdent(conflict.target.constraint)}`
+		} else {
+			const targetColumns = conflict.target
+			if (targetColumns.length === 0 || targetColumns.some((column) => !Object.hasOwn(table.columns, column))) {
+				throw new QueryBuilderDefect({ message: `${where}: the conflict target must name columns of the table` })
+			}
+			const predicate = conflict.targetWhere?.(createColumnAccessor(table.columns))
+			target = ` (${targetColumns.map(quoteIdent).join(", ")})${
+				predicate === undefined ? "" : ` WHERE ${compileSqlFragment(predicate.toFragment())}`
+			}`
+		}
+	} else if (conflict.targetWhere !== undefined) {
+		throw new QueryBuilderDefect({ message: `${where}: targetWhere needs a target` })
+	}
+	if (conflict.action === "nothing") return `\nON CONFLICT${target} DO NOTHING`
+
+	const existing = createQualifiedColumnAccessor(table.name, undefined, table.columns)
+	const excluded = createQualifiedColumnAccessor("excluded", undefined, table.columns)
+	const set = typeof conflict.set === "function" ? conflict.set(existing, excluded) : conflict.set
+	const computed = new Set<string>(table.computed ?? [])
+	// The SET record may come from data, so a bad key is a failure, as in a row.
+	const assignments = Object.entries(set as Record<string, unknown>).flatMap(([column, value]) => {
+		if (value === undefined) return []
+		if (!Object.hasOwn(table.columns, column) || computed.has(column)) {
+			throw new QueryBuilderError({
+				code: "InvalidArguments",
+				message: `${where}: onConflictDoUpdate sets ${JSON.stringify(column)}, which is not an insertable column of the table`,
+			})
+		}
+		const sql = cell(column, value, "onConflictDoUpdate set")
+		wrote(column, value, sql)
+		return [`${quoteIdent(column)} = ${sql}`]
+	})
+	if (assignments.length === 0) {
+		throw new QueryBuilderError({ code: "InvalidArguments", message: `${where}: onConflictDoUpdate sets no columns` })
+	}
+	const condition = conflict.where?.(existing, excluded)
+	return `\nON CONFLICT${target} DO UPDATE SET ${assignments.join(", ")}${
+		condition === undefined ? "" : ` WHERE ${compileSqlFragment(condition.toFragment())}`
+	}`
+}
+
 /** The RETURNING clause and its row schema, or none without `returning`. */
 const returningOf = (insert: CHInsert<any, any, any, any>) => {
 	const { table, returningFn } = insert._state
@@ -1389,52 +1457,57 @@ function compileInsert(insert: CHInsert<any, any, any, any>, params: Record<stri
 		throw new QueryBuilderError({ code: "InvalidArguments", message: `${where}: every row is empty; give at least one column` })
 	}
 
+	const dialect = currentDialect()
 	const values: Record<string, unknown> = { ...params }
 	let next = 0
-	const cell = (column: string, value: unknown, index: number): string => {
+	const cell = (column: string, value: unknown, context: string): 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}`)
+		const wire = encodeValue(table.columns[column]!.literalSchema, value, `${where}: ${context}, 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.
+	// A tenant table's insert is single-tenant when every row (and an upsert's
+	// SET, if it writes the column) 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<string>()
-	let pinned = tenant !== undefined
+	let pinned = tenant !== undefined && present.has(tenant)
+	const pin = (value: unknown, sql: string): void => {
+		if (!pinned) return
+		if (value === undefined || value === null || (isExprLike(value) && !("_paramName" in value))) pinned = false
+		else bounds.add(inlineParams(sql, values))
+	}
 
-	const tuples = withSubqueryCompiler(
+	const [tuples, conflictSql] = withSubqueryCompiler(
 		(subquery) =>
 			typeof subquery === "string" ? subquery : compileInner(subquery, values, { skipFormat: true, nested: true }).sql,
-		() =>
-			rows.map((row, index) => {
+		() => {
+			const tuples = 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))
-					}
+					const sql = cell(column, value, `row ${index}`)
+					if (column === tenant) pin(value, sql)
 					return sql
 				})
-				if (tenant !== undefined && !present.has(tenant)) pinned = false
 				return `(${cells.join(", ")})`
-			}),
+			})
+			return [tuples, onConflictClause(insert, cell, (column, value, sql) => {
+				if (column === tenant) pin(value, sql)
+			})] as const
+		},
 	)
 
-	const dialect = currentDialect()
 	const returning = returningOf(insert)
 	const returningSql =
 		returning === undefined
 			? ""
 			: `\nRETURNING ${returning.aliases.map((alias) => compileSqlFragment(aliased(returning.exprs[alias]!, alias))).join(", ")}`
 	const rendered = renderParams(
-		`INSERT INTO ${quoteIdentPath(table.name)} (${columns.map(quoteIdent).join(", ")})\nVALUES ${tuples.join(", ")}${returningSql}`,
+		`INSERT INTO ${quoteIdentPath(table.name)} (${columns.map(quoteIdent).join(", ")})\nVALUES ${tuples.join(", ")}${conflictSql}${returningSql}`,
 		values,
 		dialect,
 	)

src/ch/dialect.ts

@@ -62,6 +62,9 @@ export interface DialectClauses {
 	/** `RETURNING` after an INSERT. Absent means no: an insert with
 	 *  `.returning()` fails to compile for the dialect. */
 	readonly returning?: boolean
+	/** `ON CONFLICT ... DO NOTHING / DO UPDATE` after an INSERT. Absent means
+	 *  no: an insert with `onConflict*` fails to compile for the dialect. */
+	readonly onConflict?: boolean
 }
 
 /** A transaction isolation level. `read uncommitted` is left out: Postgres runs it as `read committed`. */
@@ -147,6 +150,7 @@ export const clickhouseDialect: Dialect = {
 		groupByAlias: true,
 		parenthesizeUnionBranches: false,
 		returning: false,
+		onConflict: false,
 	},
 	transactions: noTransactions,
 }

src/ch/index.ts

@@ -240,7 +240,17 @@ export {
 } from "./query"
 
 // Insert builder
-export { type CHInsert, type InsertRow, type InsertRowOf, type InsertValue, insertInto } from "./insert"
+export {
+	type CHInsert,
+	type ConflictSet,
+	type ConflictTarget,
+	type InsertRow,
+	type InsertRowOf,
+	type InsertValue,
+	type OnConflictDoNothing,
+	type OnConflictDoUpdate,
+	insertInto,
+} from "./insert"
 
 // Compilation
 export {

src/ch/insert.test-d.ts

@@ -94,6 +94,22 @@ expectTypeOf(PG.compileUnsafe(returningExprs)).toEqualTypeOf<
 // @ts-expect-error not a column
 CH.insertInto(Keys).values({ id: "k" }).returning("nope")
 
+// ON CONFLICT: targets are column names, SET takes values or expressions.
+const Counters = CH.table("counters", { key: PG.text, count: PG.int8 })
+CH.insertInto(Counters).values({ key: "k", count: 1 }).onConflictDoNothing({ target: ["key"] })
+CH.insertInto(Counters)
+	.values({ key: "k", count: 1 })
+	.onConflictDoUpdate({ target: ["key"], set: ($, excluded) => ({ count: $.count.add(excluded.count) }) })
+CH.insertInto(Counters).values({ key: "k", count: 1 }).onConflictDoUpdate({ target: { constraint: "c" }, set: { count: 0 } })
+// @ts-expect-error not a column
+CH.insertInto(Counters).values({ key: "k", count: 1 }).onConflictDoNothing({ target: ["nope"] })
+// @ts-expect-error a value of another type
+CH.insertInto(Counters).values({ key: "k", count: 1 }).onConflictDoUpdate({ target: ["key"], set: { count: "1" } })
+// @ts-expect-error DO UPDATE needs a target
+CH.insertInto(Counters).values({ key: "k", count: 1 }).onConflictDoUpdate({ set: { count: 0 } })
+// @ts-expect-error a computed column cannot be set
+CH.insertInto(Spans).values({ OrgId: "o", Label: "l" }).onConflictDoUpdate({ target: ["OrgId"], set: { Day: "x" } })
+
 // An insert without RETURNING runs to no rows; compile gives a CompiledQuery.
 const insert = CH.insertInto(Plain).values({ A: "a", B: 1 })
 expectTypeOf<RowOf<typeof insert>>().toEqualTypeOf<never>()