Simplify

simolus3 · simolus3 · commit 9d654b983ea1 · 2025-08-26T17:04:47.000+02:00
diff --git a/modules/module-postgres/src/replication/PgManager.ts b/modules/module-postgres/src/replication/PgManager.ts
@@ -2,7 +2,7 @@ import * as pgwire from '@powersync/service-jpgwire';
 import semver from 'semver';
 import { NormalizedPostgresConnectionConfig } from '../types/types.js';
 import { getApplicationName } from '../utils/application-name.js';
-import { PostgresTypeCache } from '../types/custom.js';
+import { PostgresTypeCache } from '../types/cache.js';
 
 /**
  * Shorter timeout for snapshot connections than for replication connections.
diff --git a/modules/module-postgres/src/replication/PgRelation.ts b/modules/module-postgres/src/replication/PgRelation.ts
@@ -30,3 +30,12 @@ export function getPgOutputRelation(source: PgoutputRelation): storage.SourceEnt
     replicaIdColumns: getReplicaIdColumns(source)
   } satisfies storage.SourceEntityDescriptor;
 }
+
+export function referencedColumnTypeIds(source: PgoutputRelation): number[] {
+  const oids = new Set<number>();
+  for (const column of source.columns) {
+    oids.add(column.typeOid);
+  }
+
+  return [...oids];
+}
diff --git a/modules/module-postgres/src/replication/WalStream.ts b/modules/module-postgres/src/replication/WalStream.ts
@@ -31,7 +31,7 @@ import {
 } from '@powersync/service-sync-rules';
 
 import { PgManager } from './PgManager.js';
-import { getPgOutputRelation, getRelId } from './PgRelation.js';
+import { getPgOutputRelation, getRelId, referencedColumnTypeIds } from './PgRelation.js';
 import { checkSourceConfiguration, checkTableRls, getReplicationIdentityColumns } from './replication-utils.js';
 import { ReplicationMetric } from '@powersync/service-types';
 import {
@@ -188,28 +188,30 @@ export class WalStream {
 
     let tableRows: any[];
     const prefix = tablePattern.isWildcard ? tablePattern.tablePrefix : undefined;
-    if (tablePattern.isWildcard) {
-      const result = await db.query({
-        statement: `SELECT c.oid AS relid, c.relname AS table_name
+
+    {
+      let query = `
+      SELECT
+        c.oid AS relid,
+        c.relname AS table_name,
+        (SELECT 
+          json_agg(DISTINCT a.atttypid)
+          FROM pg_attribute a
+          WHERE a.attnum > 0 AND NOT a.attisdropped AND a.attrelid = c.oid) 
+        AS column_types
       FROM pg_class c
       JOIN pg_namespace n ON n.oid = c.relnamespace
       WHERE n.nspname = $1
-      AND c.relkind = 'r'
-      AND c.relname LIKE $2`,
-        params: [
-          { type: 'varchar', value: schema },
-          { type: 'varchar', value: tablePattern.tablePattern }
-        ]
-      });
-      tableRows = pgwire.pgwireRows(result);
-    } else {
+      AND c.relkind = 'r'`;
+
+      if (tablePattern.isWildcard) {
+        query += ' AND c.relname LIKE $2';
+      } else {
+        query += ' AND c.relname = $2';
+      }
+
       const result = await db.query({
-        statement: `SELECT c.oid AS relid, c.relname AS table_name
-      FROM pg_class c
-      JOIN pg_namespace n ON n.oid = c.relnamespace
-      WHERE n.nspname = $1
-      AND c.relkind = 'r'
-      AND c.relname = $2`,
+        statement: query,
         params: [
           { type: 'varchar', value: schema },
           { type: 'varchar', value: tablePattern.tablePattern }
@@ -218,6 +220,7 @@ export class WalStream {
 
       tableRows = pgwire.pgwireRows(result);
     }
+
     let result: storage.SourceTable[] = [];
 
     for (let row of tableRows) {
@@ -257,16 +260,18 @@ export class WalStream {
 
       const cresult = await getReplicationIdentityColumns(db, relid);
 
-      const table = await this.handleRelation(
+      const columnTypes = (JSON.parse(row.column_types) as string[]).map((e) => Number(e));
+      const table = await this.handleRelation({
         batch,
-        {
+        descriptor: {
           name,
           schema,
           objectId: relid,
           replicaIdColumns: cresult.replicationColumns
         } as SourceEntityDescriptor,
-        false
-      );
+        snapshot: false,
+        referencedTypeIds: columnTypes
+      });
 
       result.push(table);
     }
@@ -672,7 +677,14 @@ WHERE  oid = $1::regclass`,
     }
   }
 
-  async handleRelation(batch: storage.BucketStorageBatch, descriptor: SourceEntityDescriptor, snapshot: boolean) {
+  async handleRelation(options: {
+    batch: storage.BucketStorageBatch;
+    descriptor: SourceEntityDescriptor;
+    snapshot: boolean;
+    referencedTypeIds: number[];
+  }) {
+    const { batch, descriptor, snapshot, referencedTypeIds } = options;
+
     if (!descriptor.objectId && typeof descriptor.objectId != 'number') {
       throw new ReplicationAssertionError(`objectId expected, got ${typeof descriptor.objectId}`);
     }
@@ -698,6 +710,9 @@ WHERE  oid = $1::regclass`,
       // Truncate this table, in case a previous snapshot was interrupted.
       await batch.truncate([result.table]);
 
+      // Ensure we have a description for custom types referenced in the table.
+      await this.connections.types.fetchTypes(referencedTypeIds);
+
       // Start the snapshot inside a transaction.
       // We use a dedicated connection for this.
       const db = await this.connections.snapshotConnection();
@@ -939,7 +954,12 @@ WHERE  oid = $1::regclass`,
 
           for (const msg of messages) {
             if (msg.tag == 'relation') {
-              await this.handleRelation(batch, getPgOutputRelation(msg), true);
+              await this.handleRelation({
+                batch,
+                descriptor: getPgOutputRelation(msg),
+                snapshot: true,
+                referencedTypeIds: referencedColumnTypeIds(msg)
+              });
             } else if (msg.tag == 'begin') {
               // This may span multiple transactions in the same chunk, or even across chunks.
               skipKeepalive = true;
diff --git a/modules/module-postgres/src/types/cache.ts b/modules/module-postgres/src/types/cache.ts
@@ -10,7 +10,7 @@ export class PostgresTypeCache {
   }
 
   public async fetchTypes(oids: number[]) {
-    let pending = oids.filter((id) => !(id in Object.values(pgwire.PgTypeOid)));
+    let pending = oids.filter((id) => !this.registry.knows(id));
     // For details on columns, see https://www.postgresql.org/docs/current/catalog-pg-type.html
     const statement = `
 SELECT oid, t.typtype,
diff --git a/modules/module-postgres/src/types/registry.ts b/modules/module-postgres/src/types/registry.ts
@@ -9,6 +9,7 @@ import {
   toSyncRulesValue
 } from '@powersync/service-sync-rules';
 import * as pgwire from '@powersync/service-jpgwire';
+import { JsonContainer } from '@powersync/service-jsonbig';
 
 interface BaseType {
   sqliteType: () => SqliteValueType;
@@ -167,102 +168,34 @@ export class CustomTypeRegistry {
         return pgwire.PgType.decode(raw, oid);
       case 'domain':
         return this.decodeWithCustomTypes(raw, resolved.innerId);
-    }
-
-    type StructureState = (ArrayType & { parsed: any[] }) | (CompositeType & { parsed: [string, any][] });
-    const stateStack: StructureState[] = [];
-    let pendingNestedStructure: ArrayType | CompositeType | null = resolved;
-
-    const pushParsedValue = (value: any) => {
-      const top = stateStack[stateStack.length - 1];
-      if (top.type == 'array') {
-        top.parsed.push(value);
-      } else {
-        const nextMember = top.members[top.parsed.length];
-        if (nextMember) {
-          top.parsed.push([nextMember.name, value]);
-        }
-      }
-    };
-
-    const resolveCurrentStructureTypeId = () => {
-      const top = stateStack[stateStack.length - 1];
-      if (top.type == 'array') {
-        return top.innerId;
-      } else {
-        const nextMember = top.members[top.parsed.length];
-        if (nextMember) {
-          return nextMember.typeId;
-        } else {
-          return -1;
-        }
-      }
-    };
-
-    let result: any;
-    pgwire.decodeSequence({
-      source: raw,
-      delimiters: this.delimitersFor(resolved),
-      listener: {
-        onStructureStart: () => {
-          stateStack.push({
-            ...pendingNestedStructure!,
-            parsed: []
-          });
-          pendingNestedStructure = null;
-        },
-        onValue: (raw) => {
-          pushParsedValue(raw == null ? null : this.decodeWithCustomTypes(raw, resolveCurrentStructureTypeId()));
-        },
-        onStructureEnd: () => {
-          const top = stateStack.pop()!;
-          // For arrays, pop the parsed array. For compounds, create an object from the key-value entries.
-          const parsedValue = top.type == 'array' ? top.parsed : Object.fromEntries(top.parsed);
-
-          if (stateStack.length == 0) {
-            // We have exited the outermost structure, parsedValue is the result.
-            result = parsedValue;
-          } else {
-            // Add the result of parsing a nested structure to the current outer structure.
-            pushParsedValue(parsedValue);
-          }
-        },
-        maybeParseSubStructure: (firstChar: number) => {
-          const top = stateStack[stateStack.length - 1];
-          if (top.type == 'array' && firstChar == pgwire.CHAR_CODE_LEFT_BRACE) {
-            // Postgres arrays are natively multidimensional - so if we're in an array, we can always parse sub-arrays
-            // of the same type.
-            pendingNestedStructure = top;
-            return this.delimitersFor(top);
+      case 'array':
+        return pgwire.decodeArray({
+          source: raw,
+          decodeElement: (source) => this.decodeWithCustomTypes(source, resolved.innerId),
+          delimiterCharCode: resolved.separatorCharCode
+        });
+      case 'composite': {
+        const parsed: [string, any][] = [];
+
+        pgwire.decodeSequence({
+          source: raw,
+          delimiters: pgwire.COMPOSITE_DELIMITERS,
+          listener: {
+            onValue: (raw) => {
+              const nextMember = resolved.members[parsed.length];
+              if (nextMember) {
+                const value = raw == null ? null : this.decodeWithCustomTypes(raw, nextMember.typeId);
+                parsed.push([nextMember.name, value]);
+              }
+            },
+            // These are only used for nested arrays
+            onStructureStart: () => {},
+            onStructureEnd: () => {}
           }
+        });
 
-          // If we're in a compound type, nested compound values or arrays are encoded as strings.
-          return null;
-        }
+        return Object.fromEntries(parsed);
       }
-    });
-
-    return result;
-  }
-
-  private resolveStructure(type: MaybeKnownType): [ArrayType | CompositeType, pgwire.Delimiters] | null {
-    switch (type.type) {
-      case 'builtin':
-      case 'unknown':
-        return null;
-      case 'domain':
-        return this.resolveStructure(this.lookupType(type.innerId));
-      case 'array':
-      case 'composite':
-        return [type, this.delimitersFor(type)];
-    }
-  }
-
-  private delimitersFor(type: ArrayType | CompositeType): pgwire.Delimiters {
-    if (type.type == 'array') {
-      return pgwire.arrayDelimiters(type.separatorCharCode);
-    } else {
-      return pgwire.COMPOSITE_DELIMITERS;
     }
   }
 
diff --git a/packages/jpgwire/src/pgwire_types.ts b/packages/jpgwire/src/pgwire_types.ts
@@ -1,13 +1,14 @@
 // Adapted from https://github.com/kagis/pgwire/blob/0dc927f9f8990a903f238737326e53ba1c8d094f/mod.js#L2218
 
 import { JsonContainer } from '@powersync/service-jsonbig';
-import { TimeValue, type DatabaseInputValue } from '@powersync/service-sync-rules';
+import { CustomSqliteValue, TimeValue, type DatabaseInputValue } from '@powersync/service-sync-rules';
 import { dateToSqlite, lsnMakeComparable, timestampToSqlite, timestamptzToSqlite } from './util.js';
 import {
   arrayDelimiters,
   CHAR_CODE_COMMA,
   CHAR_CODE_LEFT_BRACE,
   CHAR_CODE_RIGHT_BRACE,
+  decodeArray,
   decodeSequence,
   Delimiters,
   SequenceListener
@@ -164,43 +165,10 @@ export class PgType {
 
   static _decodeArray(text: string, elemTypeOid: number): DatabaseInputValue[] {
     text = text.replace(/^\[.+=/, ''); // skip dimensions
-
-    let results: DatabaseInputValue[];
-    const stack: DatabaseInputValue[][] = [];
-    const delimiters = arrayDelimiters();
-
-    const listener: SequenceListener = {
-      maybeParseSubStructure: function (firstChar: number): Delimiters | null {
-        return firstChar == CHAR_CODE_LEFT_BRACE ? delimiters : null;
-      },
-      onStructureStart: () => {
-        // We're parsing a new array
-        stack.push([]);
-      },
-      onValue: function (value: string | null): void {
-        // Atomic (non-array) value, add to current array.
-        stack[stack.length - 1].push(value && PgType.decode(value, elemTypeOid));
-      },
-      onStructureEnd: () => {
-        // We're done parsing an array.
-        const subarray = stack.pop()!;
-        if (stack.length == 0) {
-          // We are done with the outermost array, set results.
-          results = subarray;
-        } else {
-          // We were busy parsing a nested array, continue outer array.
-          stack[stack.length - 1].push(subarray);
-        }
-      }
-    };
-
-    decodeSequence({
+    return decodeArray({
       source: text,
-      listener,
-      delimiters
+      decodeElement: (raw) => PgType.decode(raw, elemTypeOid)
     });
-
-    return results!;
   }
 
   static _decodeBytea(text: string): Uint8Array {
diff --git a/packages/jpgwire/src/sequence_tokenizer.ts b/packages/jpgwire/src/sequence_tokenizer.ts

Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@ export class PostgresTypeCache {`
`10`	`10`	`}`
`11`	`11`
`12`	`12`	`public async fetchTypes(oids: number[]) {`
`13`		`- let pending = oids.filter((id) => !(id in Object.values(pgwire.PgTypeOid)));`
	`13`	`+ let pending = oids.filter((id) => !this.registry.knows(id));`
`14`	`14`	`// For details on columns, see https://www.postgresql.org/docs/current/catalog-pg-type.html`
`15`	`15`	const statement = `
`16`	`16`	`SELECT oid, t.typtype,`