From feaf7141fec107993d765bc8ba7793a47fdcdbbc Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 13 Oct 2026 11:34:26 -0700 Subject: [PATCH 1/4] perf(tables): keep the batch unique check on per-value GIN probes The batch unique check ORed every value into one query under LIMIT n. The LIMIT made the planner expect an early exit and walk the table's rows through a btree with every value as a filter, so large batches read the whole table inside the write transaction while holding the unique and schema locks. Split the values into UNION ALL branches of at most 25 ORed predicates with no LIMIT, so each value plans as its own GIN probe. Cast-compared columns (ttl), which no index serves, stay one pass. The predicates, null and in-batch handling, error messages, result shape, and lock order are unchanged. --- .../lib/table/column-types/comparison-sql.ts | 20 +- .../lib/table/rows/row-writes.integration.ts | 182 +++++++++++++++++- apps/sim/lib/table/sql.ts | 8 +- apps/sim/lib/table/validation.ts | 40 ++-- 4 files changed, 226 insertions(+), 24 deletions(-) diff --git a/apps/sim/lib/table/column-types/comparison-sql.ts b/apps/sim/lib/table/column-types/comparison-sql.ts index 03a9d20cea3..081b307291c 100644 --- a/apps/sim/lib/table/column-types/comparison-sql.ts +++ b/apps/sim/lib/table/column-types/comparison-sql.ts @@ -3,12 +3,20 @@ import { columnTypeOf } from '@/lib/table/column-types/registry' import { validatedTimestampSql } from '@/lib/table/column-types/timestamp-sql' import type { ColumnDefinition } from '@/lib/table/types' +/** + * The database cast an equality match on `column` compares through, or `undefined` when equality is + * JSONB containment. Only a type with an equality projection compares by cast, which no index serves. + */ +export function equalityCast(column: ColumnDefinition): string | undefined { + const definition = columnTypeOf(column) + return definition.valueForEquality ? definition.jsonbCast : undefined +} + /** Casts stored text for equality, treating malformed explicit timestamps as absent instants. */ export function columnTextForEquality(cell: SQL, column: ColumnDefinition): SQL { - const definition = columnTypeOf(column) - if (!definition.valueForEquality || !definition.jsonbCast) return cell - if (definition.timestampValidation) { - return validatedTimestampSql(cell, definition.timestampValidation) - } - return sql`(${cell})::${sql.raw(definition.jsonbCast)}` + const cast = equalityCast(column) + if (!cast) return cell + const { timestampValidation } = columnTypeOf(column) + if (timestampValidation) return validatedTimestampSql(cell, timestampValidation) + return sql`(${cell})::${sql.raw(cast)}` } diff --git a/apps/sim/lib/table/rows/row-writes.integration.ts b/apps/sim/lib/table/rows/row-writes.integration.ts index a1fdc30e4a0..93a7310287a 100644 --- a/apps/sim/lib/table/rows/row-writes.integration.ts +++ b/apps/sim/lib/table/rows/row-writes.integration.ts @@ -44,6 +44,7 @@ import { lockUniqueColumns, lockUniqueValues } from '@/lib/table/rows/unique-loc import { getTableById } from '@/lib/table/service' import { getOrCreateTableSnapshot } from '@/lib/table/snapshot-cache' import type { ColumnDefinition, JsonValue, RowData, TableDefinition } from '@/lib/table/types' +import { checkBatchUniqueConstraintsDb } from '@/lib/table/validation' const url = readTestDatabaseUrl() if (process.env.DATABASE_URL !== url) { @@ -87,7 +88,7 @@ async function createTable(columns: ColumnDefinition[]): Promise + rows: Array<{ id: string; data: RowData; orderKey: string | null; position?: number }> ) { await db.insert(userTableRows).values(rows.map((row) => ({ ...row, tableId, workspaceId }))) } @@ -481,6 +482,33 @@ describe('table row writes against real PostgreSQL', () => { expect(await storedCount(table.id, { email: 'dup@example.test' })).toBe(1) }) + it('rejects the second of two concurrent batches over the value-lock cap that share a value', async () => { + const table = await createTable(uniqueColumns) + const batch = (prefix: string) => () => + batchInsertRows( + { + tableId: table.id, + workspaceId, + rows: [ + ...Array.from({ length: 100 }, (_, i) => ({ email: `${prefix}-${i}@example.test` })), + { email: 'shared@example.test' }, + ], + secretProvenance: undefined, + capabilityGovernedUserId: null, + }, + table, + 'unique-race' + ) + + const results = await raceUnderHeldDefinitionRow(table.id, [batch('left'), batch('right')], { + onDefinitionRow: 1, + onValueLock: 1, + }) + + expect(fulfilled(results)).toBe(1) + expect(await storedCount(table.id, { email: 'shared@example.test' })).toBe(1) + }) + it('rejects an import batch racing a single insert of the same value', async () => { const table = await createTable(uniqueColumns) @@ -700,6 +728,158 @@ describe('table row writes against real PostgreSQL', () => { }) }) + describe('batch unique check across many values', () => { + /** Seeds `cells` at positions 0.., which the conflict messages report one-based. */ + const seedAt = (tableId: string, cells: RowData[]) => + seedRows( + tableId, + cells.map((data, position) => ({ + id: `${tableId}-${position}`, + data, + orderKey: `a${position}`, + position, + })) + ) + + const checkInTransaction = (table: TableDefinition, rows: RowData[]) => + db.transaction((trx) => checkBatchUniqueConstraintsDb(table.id, rows, table.schema, trx)) + + /** + * Index and sequential scans this transaction ran on `user_table_rows`, by relation, read from + * the transaction's own statistics so concurrent suites cannot skew them. + */ + async function rowScans(trx: DbTransaction): Promise> { + const scans = await trx.execute<{ relname: string; scans: number }>(sql` + SELECT c.relname, pg_stat_get_xact_numscans(c.oid)::int AS scans + FROM pg_class c + WHERE c.oid = 'user_table_rows'::regclass + OR c.oid IN (SELECT indexrelid FROM pg_index WHERE indrelid = 'user_table_rows'::regclass)`) + return Object.fromEntries( + scans.filter((row) => row.scans > 0).map((row) => [row.relname, row.scans]) + ) + } + + /** Runs the check in a transaction and returns its result with the scans it ran. */ + const checkWithScans = (table: TableDefinition, rows: RowData[]) => + db.transaction(async (trx) => { + // A parallel worker's scans are not counted in the leader's transaction statistics. + await trx.execute(sql`SET LOCAL max_parallel_workers_per_gather = 0`) + const result = await checkBatchUniqueConstraintsDb(table.id, rows, table.schema, trx) + return { result, scans: await rowScans(trx) } + }) + + it('reports stored and in-batch conflicts across every probe group', async () => { + const table = await createTable([ + { id: 'email', name: 'email', type: 'string', unique: true }, + { id: 'meta', name: 'meta', type: 'json', unique: true }, + ]) + await seedAt(table.id, [ + { email: 'first@example.test' }, + { email: 'middle@example.test' }, + { email: 'last@example.test' }, + { meta: { a: 1, b: 2 } }, + { meta: [1, 2] }, + ]) + const rows: RowData[] = Array.from({ length: 260 }, (_, i) => ({ + email: `new-${i}@example.test`, + })) + rows[0] = { email: 'new-0@example.test', meta: { b: 2, a: 1 } } + rows[1] = { email: 'new-1@example.test', meta: { a: 1 } } + rows[2] = { email: 'new-2@example.test', meta: [2, 1] } + rows[3] = { email: 'first@example.test', meta: [1, 2] } + rows[7] = { email: null } + rows[8] = {} + rows[130] = { email: 'middle@example.test' } + rows[200] = { email: 'new-5@example.test' } + rows[259] = { email: 'last@example.test' } + + expect(await checkInTransaction(table, rows)).toEqual({ + valid: false, + errors: [ + { + row: 0, + errors: [ + 'Column "meta" must be unique. Value "[object Object]" already exists in row 4', + ], + }, + { + row: 3, + errors: [ + 'Column "email" must be unique. Value "first@example.test" already exists in row 1', + 'Column "meta" must be unique. Value "1,2" already exists in row 5', + ], + }, + { + row: 130, + errors: [ + 'Column "email" must be unique. Value "middle@example.test" already exists in row 2', + ], + }, + { + row: 200, + errors: [ + 'Column "email" must be unique. Value "new-5@example.test" duplicates row 6 in batch', + ], + }, + { + row: 259, + errors: [ + 'Column "email" must be unique. Value "last@example.test" already exists in row 3', + ], + }, + ], + }) + }) + + it('matches an expiration by instant in one pass over the table', async () => { + const table = await createTable([ + { id: 'expires', name: 'expires', type: 'ttl', unique: true }, + ]) + await seedAt(table.id, [{ expires: '2024-01-31T00:00:00-00:00' }]) + const rows: RowData[] = Array.from({ length: 30 }, (_, i) => ({ + expires: `2025-01-${String(i + 1).padStart(2, '0')}T00:00:00-00:00`, + })) + rows[27] = { expires: '2024-01-31T01:00:00+01:00' } + rows[28] = { expires: null } + + const { result, scans } = await checkWithScans(table, rows) + + expect(result).toEqual({ + valid: false, + errors: [ + { + row: 27, + errors: [ + 'Column "expires" must be unique. Value "2024-01-31T01:00:00+01:00" already exists in row 1', + ], + }, + ], + }) + expect(Object.values(scans).reduce((sum, count) => sum + count, 0)).toBe(1) + }) + + it('probes the containment index once per value on a large table', async () => { + const table = await createTable([ + { id: 'email', name: 'email', type: 'string', unique: true }, + ]) + await control`INSERT INTO user_table_rows (id, table_id, workspace_id, data, position, order_key) + SELECT ${table.id}::text || '-' || i, ${table.id}, ${workspaceId}, + jsonb_build_object('email', 'stored-' || i || '@example.test'), i, 'a' || i + FROM generate_series(0, 39999) AS i` + await control`ANALYZE user_table_rows` + const rows: RowData[] = Array.from({ length: 260 }, (_, i) => ({ + email: i % 10 === 0 ? `stored-${i * 100}@example.test` : `new-${i}@example.test`, + })) + + const { result, scans } = await checkWithScans(table, rows) + + await control`DELETE FROM user_table_rows WHERE table_id = ${table.id}` + + expect(result.errors).toHaveLength(26) + expect(scans).toEqual({ user_table_rows_tenant_data_gin_idx: 260 }) + }, 60_000) + }) + describe('writers validate against the live schema, not their snapshot', () => { const columns: ColumnDefinition[] = [ { id: 'key', name: 'key', type: 'string', unique: true }, diff --git a/apps/sim/lib/table/sql.ts b/apps/sim/lib/table/sql.ts index e5a8b739515..96c3a99639b 100644 --- a/apps/sim/lib/table/sql.ts +++ b/apps/sim/lib/table/sql.ts @@ -19,7 +19,7 @@ import { SINGLE_SELECT_OPERATORS, SINGLE_SELECT_OPS, } from '@/lib/table/column-types' -import { columnTextForEquality } from '@/lib/table/column-types/comparison-sql' +import { columnTextForEquality, equalityCast } from '@/lib/table/column-types/comparison-sql' import { NAME_PATTERN } from '@/lib/table/constants' import { normalizeDateCellValue } from '@/lib/table/dates' import { TableQueryValidationError } from '@/lib/table/errors' @@ -630,10 +630,10 @@ export function fieldPredicate( : value const equalityClause = (operand: JsonValue): SQL => { - const definition = column && columnTypeOf(column) - if (column && operand !== null && definition?.valueForEquality && definition.jsonbCast) { + const cast = column && equalityCast(column) + if (column && operand !== null && cast) { const cell = columnTextForEquality(sql.raw(`${tableName}.data->>'${field}'`), column) - return sql`COALESCE(${cell} = ${operand}::${sql.raw(definition.jsonbCast)}, false)` + return sql`COALESCE(${cell} = ${operand}::${sql.raw(cast)}, false)` } return buildContainmentClause(tableName, field, operand) } diff --git a/apps/sim/lib/table/validation.ts b/apps/sim/lib/table/validation.ts index 42a8e084ba0..027f9e25394 100644 --- a/apps/sim/lib/table/validation.ts +++ b/apps/sim/lib/table/validation.ts @@ -4,6 +4,7 @@ import { db } from '@sim/db' import { userTableRows } from '@sim/db/schema' +import { chunkArray } from '@sim/utils/helpers' import { and, eq, or, type SQL, sql } from 'drizzle-orm' import { NextResponse } from 'next/server' import { canonicalJson } from '@/lib/api/cursor-binding' @@ -19,6 +20,7 @@ import { validateColumnTypeLimits, validateTypeMetadata, } from '@/lib/table/column-types' +import { equalityCast } from '@/lib/table/column-types/comparison-sql' import { getMaxRowSizeBytes, NAME_PATTERN, @@ -537,6 +539,13 @@ async function withSeqscanOffOn( */ type UniqueCheckExecutor = Pick +/** + * Most values one branch of the batch conflict query ORs together. The branches join with UNION ALL + * and carry no LIMIT, so each value stays a GIN probe: under a LIMIT the planner walks every row of + * the table instead, and a much larger OR adds a scan of the table's whole entry in that index. + */ +const UNIQUE_PROBE_GROUP_SIZE = 25 + /** * Checks unique constraints for a batch of rows using targeted database queries. * Validates both against existing database rows and within the batch itself. @@ -615,8 +624,7 @@ export async function checkBatchUniqueConstraintsDb( throw new Error(`Invalid column id: ${columnId}`) } - const valueArray = Array.from(values) - const valueConditions = valueArray.map((normalizedValue) => { + const valueConditions = Array.from(values, (normalizedValue) => { // Reconstruct the original typed value from its normalized key. Both // directions go through JSON unconditionally: keying the write on the // value's RUNTIME type while keying the read on the column's DECLARED @@ -627,20 +635,26 @@ export async function checkBatchUniqueConstraintsDb( return uniqueValuePredicate(USER_TABLE_ROWS_SQL_NAME, columnId, originalValue, column) }) - const conflictingRows = await ex - .select({ - id: userTableRows.id, - data: userTableRows.data, - position: userTableRows.position, - }) - .from(userTableRows) - .where(and(eq(userTableRows.tableId, tableId), or(...valueConditions))) - .limit(valueArray.length) // We only need up to one conflict per value + // A cast comparison is a filter over the table's rows whatever its shape, so it stays one + // pass rather than one per group. + const groups = equalityCast(column) + ? [valueConditions] + : chunkArray(valueConditions, UNIQUE_PROBE_GROUP_SIZE) + const conflictingRows = await ex.execute<{ data: RowData; position: number }>( + sql.join( + groups.map((group) => + ex + .select({ data: userTableRows.data, position: userTableRows.position }) + .from(userTableRows) + .where(and(eq(userTableRows.tableId, tableId), or(...group))) + ), + sql` UNION ALL ` + ) + ) // Map conflicts back to batch rows for (const conflict of conflictingRows) { - const conflictData = conflict.data as RowData - const conflictValue = cellOf(conflictData, columnId) + const conflictValue = cellOf(conflict.data, columnId) if (conflictValue === null || conflictValue === undefined) continue // Keyed like the batch, since stored jsonb comes back with its keys reordered. const normalizedConflictValue = uniqueValueKey(conflictValue, column) From 747404886abfb2bb2f06fb6a3b8bf62ff0fa52c9 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 13 Oct 2026 11:38:00 -0700 Subject: [PATCH 2/4] chore(tables): fold the unique-check test helpers into one --- apps/sim/lib/table/rows/row-writes.integration.ts | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/apps/sim/lib/table/rows/row-writes.integration.ts b/apps/sim/lib/table/rows/row-writes.integration.ts index 93a7310287a..d968d9f5479 100644 --- a/apps/sim/lib/table/rows/row-writes.integration.ts +++ b/apps/sim/lib/table/rows/row-writes.integration.ts @@ -741,9 +741,6 @@ describe('table row writes against real PostgreSQL', () => { })) ) - const checkInTransaction = (table: TableDefinition, rows: RowData[]) => - db.transaction((trx) => checkBatchUniqueConstraintsDb(table.id, rows, table.schema, trx)) - /** * Index and sequential scans this transaction ran on `user_table_rows`, by relation, read from * the transaction's own statistics so concurrent suites cannot skew them. @@ -759,7 +756,6 @@ describe('table row writes against real PostgreSQL', () => { ) } - /** Runs the check in a transaction and returns its result with the scans it ran. */ const checkWithScans = (table: TableDefinition, rows: RowData[]) => db.transaction(async (trx) => { // A parallel worker's scans are not counted in the leader's transaction statistics. @@ -793,7 +789,7 @@ describe('table row writes against real PostgreSQL', () => { rows[200] = { email: 'new-5@example.test' } rows[259] = { email: 'last@example.test' } - expect(await checkInTransaction(table, rows)).toEqual({ + expect((await checkWithScans(table, rows)).result).toEqual({ valid: false, errors: [ { From b21d12fbd8c3a735348c977a18a561d63a515631 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 13 Oct 2026 11:45:34 -0700 Subject: [PATCH 3/4] fix(tables): coalesce a null equality cast to undefined --- apps/sim/lib/table/column-types/comparison-sql.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/sim/lib/table/column-types/comparison-sql.ts b/apps/sim/lib/table/column-types/comparison-sql.ts index 081b307291c..e652176ae7b 100644 --- a/apps/sim/lib/table/column-types/comparison-sql.ts +++ b/apps/sim/lib/table/column-types/comparison-sql.ts @@ -9,7 +9,7 @@ import type { ColumnDefinition } from '@/lib/table/types' */ export function equalityCast(column: ColumnDefinition): string | undefined { const definition = columnTypeOf(column) - return definition.valueForEquality ? definition.jsonbCast : undefined + return definition.valueForEquality ? (definition.jsonbCast ?? undefined) : undefined } /** Casts stored text for equality, treating malformed explicit timestamps as absent instants. */ From d61989551fb0cd9e1c783a5d7508d364ecb7e79e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 13 Oct 2026 11:56:26 -0700 Subject: [PATCH 4/4] fix(tables): measure the unique-check scans by delta on a flushed GIN index --- .../lib/table/rows/row-writes.integration.ts | 23 +++++++++++++------ 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/apps/sim/lib/table/rows/row-writes.integration.ts b/apps/sim/lib/table/rows/row-writes.integration.ts index d968d9f5479..45928049e65 100644 --- a/apps/sim/lib/table/rows/row-writes.integration.ts +++ b/apps/sim/lib/table/rows/row-writes.integration.ts @@ -742,26 +742,32 @@ describe('table row writes against real PostgreSQL', () => { ) /** - * Index and sequential scans this transaction ran on `user_table_rows`, by relation, read from - * the transaction's own statistics so concurrent suites cannot skew them. + * Index and sequential scans this backend has counted on `user_table_rows`, by relation. The + * counters are the backend's unflushed statistics, which can still hold scans from earlier + * transactions on the same pooled connection, so callers diff two readings. */ - async function rowScans(trx: DbTransaction): Promise> { + async function rowScans(trx: DbTransaction): Promise> { const scans = await trx.execute<{ relname: string; scans: number }>(sql` SELECT c.relname, pg_stat_get_xact_numscans(c.oid)::int AS scans FROM pg_class c WHERE c.oid = 'user_table_rows'::regclass OR c.oid IN (SELECT indexrelid FROM pg_index WHERE indrelid = 'user_table_rows'::regclass)`) - return Object.fromEntries( - scans.filter((row) => row.scans > 0).map((row) => [row.relname, row.scans]) - ) + return new Map(scans.map((row) => [row.relname, row.scans])) } + /** Runs the check in a transaction and returns its result with the scans the check alone ran. */ const checkWithScans = (table: TableDefinition, rows: RowData[]) => db.transaction(async (trx) => { // A parallel worker's scans are not counted in the leader's transaction statistics. await trx.execute(sql`SET LOCAL max_parallel_workers_per_gather = 0`) + const before = await rowScans(trx) const result = await checkBatchUniqueConstraintsDb(table.id, rows, table.schema, trx) - return { result, scans: await rowScans(trx) } + const scans: Record = {} + for (const [relname, count] of await rowScans(trx)) { + const ran = count - (before.get(relname) ?? 0) + if (ran > 0) scans[relname] = ran + } + return { result, scans } }) it('reports stored and in-batch conflicts across every probe group', async () => { @@ -862,6 +868,9 @@ describe('table row writes against real PostgreSQL', () => { SELECT ${table.id}::text || '-' || i, ${table.id}, ${workspaceId}, jsonb_build_object('email', 'stored-' || i || '@example.test'), i, 'a' || i FROM generate_series(0, 39999) AS i` + // A bulk insert leaves its entries in the GIN pending list, which a manual ANALYZE does not + // merge and which the planner charges to every probe, so flush it for a deterministic plan. + await control`SELECT gin_clean_pending_list('user_table_rows_tenant_data_gin_idx'::regclass)` await control`ANALYZE user_table_rows` const rows: RowData[] = Array.from({ length: 260 }, (_, i) => ({ email: i % 10 === 0 ? `stored-${i * 100}@example.test` : `new-${i}@example.test`,