From 56b5461cb7cc473a52fe3c80a08f71ffe1754f62 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 22 Jun 2026 15:46:58 +0000 Subject: [PATCH] fix(pglite-sync): serialize COPY initial sync with Postgres TEXT format MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The COPY-based initial insert path (`initialInsertMethod: 'csv'` / the deprecated `useCopy`) hand-rolled a CSV encoder that broke on many column types: arrays were emitted via `Array.toString` as `1,2,3` instead of the `{1,2,3}` literal, `json`/`jsonb` objects became `[object Object]`, and embedded delimiters, newlines and backslashes were mishandled. Replace it with a faithful port of PostgreSQL's own COPY TEXT serialization — `CopyAttributeOutText` (copyto.c) for field escaping and `array_out` (arrayfuncs.c) for array literals — driven by the JS types the Electric client produces. Column types are looked up from `information_schema` to disambiguate `json`/`jsonb` (whose parsed values are indistinguishable from SQL arrays by runtime type alone). Adds unit tests plus round-trip tests that push the generated stream through a real PGlite `COPY ... FROM` for scalars, strings with special characters, arrays (incl. multi-dimensional), json/jsonb (incl. `jsonb[]`), bytea and timestamps. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01VkNJeV2Ju2TnjxSxSzgL9i --- .changeset/pglite-sync-copy-text-format.md | 5 + packages/pglite-sync/src/apply.ts | 54 +++-- packages/pglite-sync/src/copy.ts | 220 +++++++++++++++++ packages/pglite-sync/test/copy.test.ts | 264 +++++++++++++++++++++ 4 files changed, 520 insertions(+), 23 deletions(-) create mode 100644 .changeset/pglite-sync-copy-text-format.md create mode 100644 packages/pglite-sync/src/copy.ts create mode 100644 packages/pglite-sync/test/copy.test.ts diff --git a/.changeset/pglite-sync-copy-text-format.md b/.changeset/pglite-sync-copy-text-format.md new file mode 100644 index 000000000..a577f6f31 --- /dev/null +++ b/.changeset/pglite-sync-copy-text-format.md @@ -0,0 +1,5 @@ +--- +'@electric-sql/pglite-sync': patch +--- + +Fix the COPY-based initial sync (`initialInsertMethod: 'csv'` / `useCopy`) to serialize rows using PostgreSQL's own COPY TEXT format. The previous ad-hoc CSV encoding broke on many types (arrays became `1,2,3` instead of `{1,2,3}`, `json`/`jsonb` became `[object Object]`, and embedded delimiters/newlines/backslashes were mishandled). The new serializer is a faithful port of the backend's `CopyAttributeOutText` (copyto.c) and `array_out` (arrayfuncs.c) routines, so booleans, numbers, bigints, arrays (incl. multi-dimensional), `json`/`jsonb`, `bytea`, timestamps and arbitrary strings now round-trip correctly through `COPY ... FROM`. diff --git a/packages/pglite-sync/src/apply.ts b/packages/pglite-sync/src/apply.ts index 024f968d3..724df0d4d 100644 --- a/packages/pglite-sync/src/apply.ts +++ b/packages/pglite-sync/src/apply.ts @@ -1,6 +1,7 @@ import { ChangeMessage } from '@electric-sql/client' import type { PGliteInterface, Transaction } from '@electric-sql/pglite' import type { MapColumns, InsertChangeMessage } from './types' +import { generateCopyData } from './copy' export interface ApplyMessageToTableOptions { pg: PGliteInterface | Transaction @@ -294,6 +295,8 @@ export async function applyMessagesToTableWithCopy({ }: BulkApplyMessagesToTableOptions) { if (debug) console.log('applying messages with COPY') + if (messages.length === 0) return + // Map the messages to the data to be inserted const data: Record[] = messages.map((message) => mapColumns ? doMapColumns(mapColumns, message) : message.value, @@ -302,36 +305,41 @@ export async function applyMessagesToTableWithCopy({ // Get column names from the first message const columns = Object.keys(data[0]) - // Create CSV data - const csvData = data - .map((message) => { - return columns - .map((column) => { - const value = message[column] - // Escape double quotes and wrap in quotes if necessary - if ( - typeof value === 'string' && - (value.includes(',') || value.includes('"') || value.includes('\n')) - ) { - return `"${value.replace(/"/g, '""')}"` - } - return value === null ? '\\N' : value - }) - .join(',') - }) - .join('\n') - const csvBlob = new Blob([csvData], { type: 'text/csv' }) - - // Perform COPY FROM + // Look up the target column types. This is only needed to disambiguate + // json/jsonb columns (whose parsed values can be plain JS arrays/objects, + // indistinguishable from SQL arrays by runtime type alone). + const columnTypes = Object.fromEntries( + ( + await pg.query<{ column_name: string; udt_name: string }>( + ` + SELECT column_name, udt_name + FROM information_schema.columns + WHERE table_name = $1 AND table_schema = $2 + `, + [table, schema], + ) + ).rows.map((c) => [c.column_name, c.udt_name]), + ) + + // Serialize the rows using Postgres' own COPY TEXT format. This faithfully + // reproduces the backend's CopyAttributeOutText / array_out algorithms, so + // every built-in type — booleans, numbers, arrays, json/jsonb, timestamps, + // bytea, strings with embedded delimiters/newlines, etc. — round-trips + // correctly (the previous ad-hoc CSV encoding did not). + const copyData = generateCopyData(data, columns, columnTypes) + const copyBlob = new Blob([copyData], { type: 'text/plain' }) + + // Perform COPY FROM. TEXT is the default format; its default delimiter is a + // tab and its default NULL marker is \N, both of which generateCopyData uses. await pg.query( ` COPY "${schema}"."${table}" (${columns.map((c) => `"${c}"`).join(', ')}) FROM '/dev/blob' - WITH (FORMAT csv, NULL '\\N') + WITH (FORMAT text) `, [], { - blob: csvBlob, + blob: copyBlob, }, ) diff --git a/packages/pglite-sync/src/copy.ts b/packages/pglite-sync/src/copy.ts new file mode 100644 index 000000000..a93c284b8 --- /dev/null +++ b/packages/pglite-sync/src/copy.ts @@ -0,0 +1,220 @@ +/** + * Serialization of JavaScript values into a PostgreSQL `COPY ... WITH (FORMAT + * text)` stream. + * + * The Electric client parses the wire values it receives into native JS values + * before they reach this module: `int2`/`int4`/`float4`/`float8` become + * `number`, `int8` becomes `bigint`, `bool` becomes `boolean`, `json`/`jsonb` + * become parsed objects/arrays, array columns become (possibly nested) JS + * arrays, and every other type is left as its raw Postgres text representation + * (a `string`). To feed those values back into `COPY` we have to reverse that: + * turn each value into the exact text Postgres' input functions expect, then + * apply the COPY framing. + * + * Rather than invent an escaping scheme (the previous CSV-based approach broke + * on arrays, JSON, embedded delimiters, etc.) this is a faithful port of the + * two relevant PostgreSQL backend routines: + * + * - `CopyAttributeOutText` (src/backend/commands/copyto.c) — field escaping + * - `array_out` (src/backend/utils/adt/arrayfuncs.c) — array literals + * + * The TEXT format is used (not CSV) because it is what Postgres itself emits + * internally, has a single well-defined escaping algorithm, and round-trips + * every built-in type. + */ + +// Defaults for COPY ... WITH (FORMAT text), matching the Postgres backend. +const DELIMITER = '\t' +const NULL_MARKER = '\\N' +const ROW_SEPARATOR = '\n' + +// CopyAttributeOutText escapes the backslash, the field delimiter, and the +// control characters that have C-style escapes. Every other byte is emitted +// literally. Our delimiter is a tab, which already has a C-style escape, so the +// single map below covers all cases. +const COPY_TEXT_ESCAPES: Record = { + '\\': '\\\\', + '\b': '\\b', + '\f': '\\f', + '\n': '\\n', + '\r': '\\r', + '\t': '\\t', + '\v': '\\v', +} +const COPY_TEXT_ESCAPE_RE = /[\\\b\f\n\r\t\v]/g + +/** + * Escape an already-stringified field value per `CopyAttributeOutText`. + */ +function escapeCopyText(value: string): string { + return value.replace(COPY_TEXT_ESCAPE_RE, (c) => COPY_TEXT_ESCAPES[c]) +} + +/** + * Convert a Uint8Array to Postgres `bytea` hex-format text (`\xDEADBEEF`). + * Electric normally delivers `bytea` already as such a string, so this only + * matters for callers using a custom parser that yields binary. + */ +function byteaToText(bytes: Uint8Array): string { + let hex = '' + for (let i = 0; i < bytes.length; i++) { + hex += bytes[i].toString(16).padStart(2, '0') + } + return '\\x' + hex +} + +/** + * Render a JS `number` the way Postgres' float/int input accepts it. The only + * special cases are the non-finite values, which Postgres spells out in words. + */ +function numberToText(value: number): string { + if (Number.isNaN(value)) return 'NaN' + if (value === Infinity) return 'Infinity' + if (value === -Infinity) return '-Infinity' + return String(value) +} + +// array_out quotes an element when it is empty, looks like the literal NULL, or +// contains a brace, the element delimiter (comma), a double-quote, a backslash, +// or ASCII whitespace. We mirror `array_isspace`, which is stricter than JS +// `\s` (it excludes Unicode whitespace), to stay byte-for-byte compatible. +const ARRAY_NEEDS_QUOTE_RE = /[{}",\\ \t\n\r\v\f]/ + +/** + * Quote/escape a single array element's text per `array_out`. + */ +function quoteArrayElement(text: string): string { + const needsQuote = + text.length === 0 || + text.toLowerCase() === 'null' || + ARRAY_NEEDS_QUOTE_RE.test(text) + if (!needsQuote) return text + // Inside quotes only `"` and `\` are escaped, each with a single backslash. + return '"' + text.replace(/(["\\])/g, '\\$1') + '"' +} + +/** + * Build a Postgres array literal (`{...}`) from a JS array, recursing into + * nested arrays for multi-dimensional arrays. NULL elements become an unquoted + * `NULL`; nested arrays are emitted as bare `{...}` (never quoted), exactly as + * `array_out` does. + */ +function arrayToText(arr: ReadonlyArray): string { + const elements = arr.map((el) => { + if (el === null || el === undefined) return 'NULL' + if (Array.isArray(el)) return arrayToText(el) + return quoteArrayElement(valueToText(el)) + }) + return '{' + elements.join(',') + '}' +} + +/** + * Convert a non-null JS value to its bare Postgres text representation (before + * COPY field escaping is applied). Dispatch is on the runtime type produced by + * the Electric parser. `json`/`jsonb` columns are handled ahead of this by + * `serializeCopyValue` when a column type is known; reaching the object branch + * here is the type-less fallback. + */ +function valueToText(value: unknown): string { + switch (typeof value) { + case 'string': + return value + case 'number': + return numberToText(value) + case 'bigint': + return value.toString() + case 'boolean': + return value ? 't' : 'f' + case 'object': { + if (Array.isArray(value)) return arrayToText(value) + if (value instanceof Date) return value.toISOString() + if (value instanceof Uint8Array) return byteaToText(value) + if (value instanceof ArrayBuffer) + return byteaToText(new Uint8Array(value)) + if (ArrayBuffer.isView(value)) { + const v = value as ArrayBufferView + return byteaToText(new Uint8Array(v.buffer, v.byteOffset, v.byteLength)) + } + // json / jsonb arrive already parsed; re-serialize them. + return JSON.stringify(value) + } + default: + // Should be unreachable for Electric-parsed values; be defensive. + return String(value) + } +} + +/** + * Build a Postgres array literal from a `json[]`/`jsonb[]` value. Each element + * is JSON text (re-serialized from the parsed value), so we cannot reuse the + * generic array path: a JSON array element must stay JSON (`[1,2]`), not be + * turned into a nested SQL array (`{1,2}`). + */ +function jsonArrayToText(arr: ReadonlyArray): string { + const elements = arr.map((el) => + el === null || el === undefined + ? 'NULL' + : quoteArrayElement(jsonToText(el)), + ) + return '{' + elements.join(',') + '}' +} + +/** + * Re-serialize a parsed `json`/`jsonb` value to its JSON text form. Electric + * delivers these columns already run through `JSON.parse`, so the value is the + * decoded JS value (object, array, string, number, boolean or null) and always + * needs `JSON.stringify` to become valid JSON input again — including scalars + * (the string `hi` must be written as `"hi"`). + */ +function jsonToText(value: unknown): string { + return JSON.stringify(value) +} + +/** + * Postgres `udt_name`s that need JSON-aware serialization rather than the + * generic runtime-type dispatch. Used to resolve the otherwise undecidable + * "is this JS array a SQL array or a JSON array?" ambiguity. + */ +const JSON_UDT_NAMES = new Set(['json', 'jsonb']) +const JSON_ARRAY_UDT_NAMES = new Set(['_json', '_jsonb']) + +/** + * Serialize a single value into one COPY TEXT field. `null`/`undefined` become + * the NULL marker (`\N`); everything else is converted to text and escaped. + * + * `udtName` is the column's Postgres `udt_name` (from `information_schema`). + * When supplied it disambiguates `json`/`jsonb` columns; when omitted the + * value's runtime type is used (which cannot tell a `jsonb` array from a SQL + * array). + */ +export function serializeCopyValue(value: unknown, udtName?: string): string { + if (value === null || value === undefined) return NULL_MARKER + if (udtName !== undefined) { + if (JSON_UDT_NAMES.has(udtName)) return escapeCopyText(jsonToText(value)) + if (JSON_ARRAY_UDT_NAMES.has(udtName) && Array.isArray(value)) + return escapeCopyText(jsonArrayToText(value)) + } + return escapeCopyText(valueToText(value)) +} + +/** + * Serialize a list of row objects into the body of a `COPY ... FROM` request in + * TEXT format. Columns are emitted in the given order, fields are tab + * separated, and rows are newline separated. + * + * `columnTypes` optionally maps a column name to its Postgres `udt_name`; pass + * it so `json`/`jsonb` columns serialize correctly. + */ +export function generateCopyData( + rows: ReadonlyArray>, + columns: ReadonlyArray, + columnTypes?: Readonly>, +): string { + return rows + .map((row) => + columns + .map((column) => serializeCopyValue(row[column], columnTypes?.[column])) + .join(DELIMITER), + ) + .join(ROW_SEPARATOR) +} diff --git a/packages/pglite-sync/test/copy.test.ts b/packages/pglite-sync/test/copy.test.ts new file mode 100644 index 000000000..6901b6d10 --- /dev/null +++ b/packages/pglite-sync/test/copy.test.ts @@ -0,0 +1,264 @@ +import { PGlite } from '@electric-sql/pglite' +import { describe, expect, it } from 'vitest' +import { generateCopyData, serializeCopyValue } from '../src/copy.js' + +describe('COPY TEXT serializer', () => { + describe('serializeCopyValue', () => { + it('renders null/undefined as the \\N marker', () => { + expect(serializeCopyValue(null)).toBe('\\N') + expect(serializeCopyValue(undefined)).toBe('\\N') + }) + + it('renders booleans as t/f', () => { + expect(serializeCopyValue(true)).toBe('t') + expect(serializeCopyValue(false)).toBe('f') + }) + + it('renders numbers and bigints', () => { + expect(serializeCopyValue(42)).toBe('42') + expect(serializeCopyValue(-3.14)).toBe('-3.14') + expect(serializeCopyValue(0)).toBe('0') + expect(serializeCopyValue(123456789012345678901234567890n)).toBe( + '123456789012345678901234567890', + ) + expect(serializeCopyValue(NaN)).toBe('NaN') + expect(serializeCopyValue(Infinity)).toBe('Infinity') + expect(serializeCopyValue(-Infinity)).toBe('-Infinity') + }) + + it('escapes control characters and backslashes per CopyAttributeOutText', () => { + expect(serializeCopyValue('a\tb')).toBe('a\\tb') + expect(serializeCopyValue('a\nb')).toBe('a\\nb') + expect(serializeCopyValue('a\rb')).toBe('a\\rb') + expect(serializeCopyValue('back\\slash')).toBe('back\\\\slash') + expect(serializeCopyValue('\b\f\v')).toBe('\\b\\f\\v') + // A literal "\N" must not be mistaken for the NULL marker. + expect(serializeCopyValue('\\N')).toBe('\\\\N') + }) + + it('leaves ordinary strings untouched', () => { + expect(serializeCopyValue('hello world')).toBe('hello world') + expect(serializeCopyValue('2021-01-01 00:00:00')).toBe( + '2021-01-01 00:00:00', + ) + }) + + it('serializes 1-D arrays as Postgres array literals', () => { + expect(serializeCopyValue([1, 2, 3])).toBe('{1,2,3}') + expect(serializeCopyValue([true, false])).toBe('{t,f}') + expect(serializeCopyValue([])).toBe('{}') + }) + + it('quotes array elements that need it (array_out rules)', () => { + // empty string, comma, whitespace, braces, the literal NULL, quotes + expect(serializeCopyValue(['', 'a,b', 'a b'])).toBe('{"","a,b","a b"}') + expect(serializeCopyValue(['NULL', 'null', 'Null'])).toBe( + '{"NULL","null","Null"}', + ) + // A backslash inside an element is escaped once for array_out and again + // for the COPY field: \ -> \\ -> \\\\ + expect(serializeCopyValue(['a"b'])).toBe('{"a\\\\"b"}') + }) + + it('represents NULL array elements as unquoted NULL', () => { + expect(serializeCopyValue([1, null, 3])).toBe('{1,NULL,3}') + }) + + it('serializes nested (multi-dimensional) arrays', () => { + expect( + serializeCopyValue([ + [1, 2], + [3, 4], + ]), + ).toBe('{{1,2},{3,4}}') + }) + + it('serializes parsed json/jsonb values back to JSON text', () => { + expect(serializeCopyValue({ a: 1, b: [2, 3] })).toBe('{"a":1,"b":[2,3]}') + }) + + it('renders Uint8Array as bytea hex', () => { + expect(serializeCopyValue(new Uint8Array([0xde, 0xad, 0xbe, 0xef]))).toBe( + '\\\\xdeadbeef', + ) + }) + }) + + describe('generateCopyData', () => { + it('joins fields with tabs and rows with newlines', () => { + const rows = [ + { id: 1, name: 'a', done: true }, + { id: 2, name: 'b', done: false }, + ] + expect(generateCopyData(rows, ['id', 'name', 'done'])).toBe( + '1\ta\tt\n2\tb\tf', + ) + }) + }) + + // The strongest guarantee: feed the generated stream straight into a real + // Postgres COPY and confirm every value round-trips. + describe('round-trips through PGlite COPY FROM', () => { + async function roundTrip( + columnsDdl: string, + columns: string[], + rows: Record[], + ): Promise[]> { + const pg = await PGlite.create() + await pg.exec(`CREATE TABLE t (${columnsDdl});`) + + const columnTypes = Object.fromEntries( + ( + await pg.query<{ column_name: string; udt_name: string }>( + `SELECT column_name, udt_name FROM information_schema.columns + WHERE table_name = 't'`, + ) + ).rows.map((c) => [c.column_name, c.udt_name]), + ) + + const copyData = generateCopyData(rows, columns, columnTypes) + const blob = new Blob([copyData], { type: 'text/plain' }) + await pg.query( + `COPY t (${columns.map((c) => `"${c}"`).join(', ')}) + FROM '/dev/blob' WITH (FORMAT text)`, + [], + { blob }, + ) + + const result = await pg.query>( + `SELECT * FROM t ORDER BY id`, + ) + await pg.close() + return result.rows + } + + it('round-trips scalar types', async () => { + const rows = [ + { + id: 1, + flag: true, + n: 42, + big: 9007199254740993n, + amount: '123.45', + note: 'plain', + }, + { + id: 2, + flag: false, + n: -7, + big: -1n, + amount: '0.001', + note: null, + }, + ] + const out = await roundTrip( + 'id int, flag boolean, n int, big bigint, amount numeric, note text', + ['id', 'flag', 'n', 'big', 'amount', 'note'], + rows, + ) + // pglite returns int8 as bigint only when outside the safe-integer range, + // otherwise as a number. + expect(out).toEqual([ + { + id: 1, + flag: true, + n: 42, + big: 9007199254740993n, + amount: '123.45', + note: 'plain', + }, + { id: 2, flag: false, n: -7, big: -1, amount: '0.001', note: null }, + ]) + }) + + it('round-trips strings with delimiters, newlines and backslashes', async () => { + const rows = [ + { id: 1, s: 'tab\there' }, + { id: 2, s: 'new\nline' }, + { id: 3, s: 'carriage\rreturn' }, + { id: 4, s: 'back\\slash and "quote", comma' }, + { id: 5, s: '\\N is not null' }, + { id: 6, s: '' }, + ] + const out = await roundTrip('id int, s text', ['id', 's'], rows) + expect(out).toEqual(rows) + }) + + it('round-trips array columns', async () => { + const rows = [ + { id: 1, ints: [1, 2, 3], texts: ['a', 'b,c', 'd e'] }, + { id: 2, ints: [], texts: ['NULL', '', 'quote"here'] }, + { id: 3, ints: [10, null, 30], texts: null }, + ] + const out = await roundTrip( + 'id int, ints int[], texts text[]', + ['id', 'ints', 'texts'], + rows, + ) + expect(out).toEqual(rows) + }) + + it('round-trips multi-dimensional arrays', async () => { + const rows = [ + { + id: 1, + grid: [ + [1, 2], + [3, 4], + ], + }, + ] + const out = await roundTrip('id int, grid int[][]', ['id', 'grid'], rows) + expect(out).toEqual(rows) + }) + + it('round-trips json and jsonb', async () => { + const rows = [ + { id: 1, doc: { a: 1, b: ['x', 'y'], c: null } }, + { id: 2, doc: [1, 2, { nested: true }] }, + ] + const out = await roundTrip('id int, doc jsonb', ['id', 'doc'], rows) + expect(out).toEqual(rows) + }) + + it('round-trips json/jsonb arrays (json[]) with mixed element shapes', async () => { + const rows = [ + { id: 1, docs: [{ a: 1 }, [1, 2, 3], 'str', 42, true] }, + { id: 2, docs: null }, + ] + const out = await roundTrip('id int, docs jsonb[]', ['id', 'docs'], rows) + expect(out).toEqual(rows) + }) + + it('round-trips timestamps and uuids delivered as strings', async () => { + const rows = [ + { + id: 1, + ts: '2021-01-01 12:34:56', + uid: '00000000-0000-0000-0000-000000000001', + }, + ] + const out = await roundTrip( + 'id int, ts timestamp, uid uuid', + ['id', 'ts', 'uid'], + rows, + ) + // pglite parses timestamps into Date objects; avoid asserting an exact + // instant here since timestamp-without-tz parsing is locale dependent. + expect(out[0].uid).toBe('00000000-0000-0000-0000-000000000001') + expect(out[0].ts).toBeInstanceOf(Date) + }) + + it('round-trips bytea from a hex string and from Uint8Array', async () => { + const rows = [ + { id: 1, b: '\\xdeadbeef' }, + { id: 2, b: new Uint8Array([0x01, 0x02, 0x03]) }, + ] + const out = await roundTrip('id int, b bytea', ['id', 'b'], rows) + expect(out).toEqual([ + { id: 1, b: new Uint8Array([0xde, 0xad, 0xbe, 0xef]) }, + { id: 2, b: new Uint8Array([0x01, 0x02, 0x03]) }, + ]) + }) + }) +})