From 585a1358cde240cda4d8f2cfedf4c4af79a9c9a5 Mon Sep 17 00:00:00 2001 From: John Pruitt Date: Fri, 4 Sep 2026 09:47:16 -0500 Subject: [PATCH] fix: serialize extension arrays after provisioning --- CHANGELOG.md | 8 ++ packages/core/src/write.ts | 24 +++- ...fresh-extension-arrays.integration.test.ts | 135 ++++++++++++++++++ packages/core/test/support/db.ts | 7 +- 4 files changed, 165 insertions(+), 9 deletions(-) create mode 100644 packages/core/test/fresh-extension-arrays.integration.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 75bc8cd..6170fbb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- `upsert` and `upsertMany` now work on the same `postgres.js` pool connection + that `createIndex` used to install `ltree` and `pgvector`. Batch arrays are + bound through the built-in `text[]` serializer and cast to their qualified + destination types, avoiding stale per-connection custom-array type maps while + preserving parameterization, nulls, and array escaping. + ## [0.1.0] - 2026-09-03 ### Added diff --git a/packages/core/src/write.ts b/packages/core/src/write.ts index 2d1faed..d7ecdff 100644 --- a/packages/core/src/write.ts +++ b/packages/core/src/write.ts @@ -14,6 +14,8 @@ import { runSql } from "./sql/exec.ts"; import { normalizeTemporalTuple, temporalTupleSchema } from "./temporal.ts"; const MAX_UPSERT_BATCH_SIZE = 1000; +/** Built-in PostgreSQL `text` OID, used as sql.array's element type. */ +const POSTGRES_TEXT_OID = 25; const recordSchema = z .object({ @@ -141,17 +143,27 @@ export async function upsertMany( if (ids.length === 0) return []; const { sql } = index; + // postgres.js discovers array serializers when each connection starts. If + // createIndex installs extension types on that same connection, its cached + // map does not yet contain ltree[] or vector[]. Bind every parallel array as + // the always-known text[] type, then let PostgreSQL cast each element to the + // routine's qualified destination type. sql.array owns array escaping, nulls, + // and empty-array encoding; values remain parameters. + const embeddingArray = + index.vectorType === "halfvec" + ? sql`${sql.array(embeddings, POSTGRES_TEXT_OID)}::text[]::public.halfvec[]` + : sql`${sql.array(embeddings, POSTGRES_TEXT_OID)}::text[]::public.vector[]`; const rows = await runBatchUpsert( sql` select ord, id, status from ${sql(index.schema)}.batch_upsert - ( ${ids} - , ${contents} + ( ${sql.array(ids, POSTGRES_TEXT_OID)}::text[]::uuid[] + , ${sql.array(contents, POSTGRES_TEXT_OID)}::text[] , ${sql.json(metas)} - , ${trees} - , ${temporals} - , ${names} - , ${embeddings} + , ${sql.array(trees, POSTGRES_TEXT_OID)}::text[]::public.ltree[] + , ${sql.array(temporals, POSTGRES_TEXT_OID)}::text[]::tstzrange[] + , ${sql.array(names, POSTGRES_TEXT_OID)}::text[] + , ${embeddingArray} , ${parsedOptions.onConflict} ) `, diff --git a/packages/core/test/fresh-extension-arrays.integration.test.ts b/packages/core/test/fresh-extension-arrays.integration.test.ts new file mode 100644 index 0000000..a36e42a --- /dev/null +++ b/packages/core/test/fresh-extension-arrays.integration.test.ts @@ -0,0 +1,135 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import type { Sql } from "postgres"; +import { createIndex } from "../src/create-index.ts"; +import { openIndex } from "../src/open-index.ts"; +import { + connect, + connectToDatabase, + createTestDatabase, + dropTestDatabase, + randomTestDatabase, +} from "./support/db.ts"; + +/** + * Regression: postgres.js caches custom array serializers when a connection + * starts. createIndex may install ltree/vector after that discovery has run, so + * a bare JS array sent on the same connection degrades to `a,b` instead of a + * PostgreSQL array literal. A one-connection pool makes that sequence + * deterministic and also proves a prepared statement can be reused. + */ +test("batch writes work when createIndex installs extension array types on the same connection", async () => { + const admin = connect(); + const database = randomTestDatabase(); + let indexSql: Sql | undefined; + + try { + await createTestDatabase(admin, database); + indexSql = connectToDatabase(database, 1); + + // Open the only pool connection before the extension types exist. This + // freezes postgres.js's initial array-type map without ltree[]/halfvec[]. + const installed = await indexSql<{ readonly extname: string }[]>` + select extname + from pg_catalog.pg_extension + where extname in ('vector', 'pg_textsearch', 'ltree') + `; + assert.deepEqual(Array.from(installed), []); + + await createIndex(indexSql, "docs", { dimensions: 4 }); + const index = await openIndex(indexSql, "docs", { + embedding: "mock-embedding", + }); + + const first = await index.upsertMany([ + { + content: "Auth tokens rotate daily.", + tree: "docs.auth", + temporal: ["2026-01-01T00:00:00Z"], + embedding: [1, 0, 0, 0], + }, + { + content: "Rate limits apply per API key.", + tree: "docs.api", + name: "rate-limit", + temporal: ["2026-02-01T00:00:00Z", "2026-03-01T00:00:00Z"], + }, + { + content: "Backups are retained for thirty days.", + tree: "docs.ops", + }, + ]); + assert.deepEqual( + first.map((result) => result.status), + ["inserted", "inserted", "inserted"], + ); + + // Reuse the prepared batch statement and exercise text-array escaping. + const second = await index.upsertMany([ + { + content: 'Cache keys may contain commas, quotes " and backslashes \\.', + tree: "docs.cache", + name: 'key,"primary\\name', + embedding: [0, 1, 0, 0], + }, + ]); + assert.equal(second[0]?.status, "inserted"); + + const rows = await indexSql< + { + readonly content: string; + readonly tree: string; + readonly name: string | null; + readonly temporal: string | null; + readonly embedding: string | null; + }[] + >` + select content, tree::text as tree, name, temporal, embedding + from docs.record + order by tree + `; + assert.deepEqual( + rows.map((row) => row.tree), + ["docs.api", "docs.auth", "docs.cache", "docs.ops"], + ); + assert.equal(rows[0]?.name, "rate-limit"); + assert.ok(rows[0]?.temporal); + assert.equal(rows[0]?.embedding, null); + assert.equal(rows[1]?.name, null); + assert.ok(rows[1]?.temporal); + assert.equal(rows[1]?.embedding, "[1,0,0,0]"); + assert.equal(rows[2]?.name, 'key,"primary\\name'); + assert.equal( + rows[2]?.content, + 'Cache keys may contain commas, quotes " and backslashes \\.', + ); + assert.equal(rows[2]?.embedding, "[0,1,0,0]"); + assert.equal(rows[3]?.name, null); + assert.equal(rows[3]?.temporal, null); + assert.equal(rows[3]?.embedding, null); + + // The alternate pgvector storage type uses a distinct array cast path. + await createIndex(indexSql, "docs_vector", { + dimensions: 4, + vectorType: "vector", + }); + const vectorIndex = await openIndex(indexSql, "docs_vector", { + embedding: "mock-embedding", + }); + const vectorResult = await vectorIndex.upsert({ + content: "Full precision vector.", + tree: "docs.vectors", + embedding: [0, 0, 1, 0], + }); + const [vectorRow] = await indexSql<{ readonly embedding: string | null }[]>` + select embedding + from docs_vector.record + where id = ${vectorResult.id} + `; + assert.equal(vectorRow?.embedding, "[0,0,1,0]"); + } finally { + await indexSql?.end(); + await dropTestDatabase(admin, database); + await admin.end(); + } +}); diff --git a/packages/core/test/support/db.ts b/packages/core/test/support/db.ts index a30f3ed..3c0aa6d 100644 --- a/packages/core/test/support/db.ts +++ b/packages/core/test/support/db.ts @@ -8,15 +8,16 @@ export function connect(): Sql { return connectToUrl(testDatabaseUrl()); } -export function connectToDatabase(database: string): Sql { +export function connectToDatabase(database: string, max?: number): Sql { const url = new URL(testDatabaseUrl()); url.pathname = `/${database}`; - return connectToUrl(url.toString()); + return connectToUrl(url.toString(), max); } -function connectToUrl(url: string): Sql { +function connectToUrl(url: string, max?: number): Sql { return postgres(url, { onnotice: () => {}, + ...(max === undefined ? {} : { max }), }); }