Add test for changing schema

Author: simolus3

modules/module-postgres/src/replication/WalStream.ts

@@ -711,7 +711,7 @@ WHERE  oid = $1::regclass`,
     await batch.drop(result.dropTables);
 
     // Ensure we have a description for custom types referenced in the table.
-    //await this.connections.types.fetchTypes(referencedTypeIds);
+    await this.connections.types.fetchTypes(referencedTypeIds);
 
     // Snapshot if:
     // 1. Snapshot is requested (false for initial snapshot, since that process handles it elsewhere)

modules/module-postgres/test/src/schema_changes.test.ts

@@ -1,6 +1,6 @@
 import { compareIds, putOp, reduceBucket, removeOp, test_utils } from '@powersync/service-core-tests';
 import * as timers from 'timers/promises';
-import { describe, expect, test, it } from 'vitest';
+import { describe, expect, test } from 'vitest';
 
 import { storage } from '@powersync/service-core';
 import { describeWithStorage } from './util.js';
@@ -591,23 +591,35 @@ function defineTests(factory: storage.TestStorageFactory) {
     expect(failures).toEqual([]);
   });
 
-  it.skip('custom types', async () => {
+  test('custom types', async () => {
     await using context = await WalStreamTestContext.open(factory);
 
     await context.updateSyncRules(`
 streams:
   stream:
-    query: SELECT id, * FROM "test_data"
+    query: SELECT * FROM "test_data"
 
 config:
   edition: 2
 `);
 
     const { pool } = context;
-    await pool.query(`DROP TABLE IF EXISTS test_data`);
-    await pool.query(`CREATE TABLE test_data(id text primary key, description composite);`);
-    await pool.query(`INSERT INTO test_data(id, description) VALUES ('t1', ROW(TRUE, 2)::composite)`);
+    await pool.query(`CREATE TABLE test_data(id text primary key);`);
+    await pool.query(`INSERT INTO test_data(id) VALUES ('t1')`);
 
     await context.replicateSnapshot();
+    context.startStreaming();
+
+    await pool.query(
+      { statement: `CREATE TYPE composite AS (foo bool, bar int4);` },
+      { statement: `ALTER TABLE test_data ADD COLUMN other composite;` },
+      { statement: `UPDATE test_data SET other = ROW(TRUE, 2)::composite;` }
+    );
+
+    const data = await context.getBucketData('1#stream|0[]');
+    expect(data).toMatchObject([
+      putOp('test_data', { id: 't1' }),
+      putOp('test_data', { id: 't1', other: '{"foo":1,"bar":2}' })
+    ]);
   });
 }

modules/module-postgres/test/src/wal_stream.test.ts

@@ -4,7 +4,7 @@ import { METRICS_HELPER, putOp, removeOp } from '@powersync/service-core-tests';
 import { pgwireRows } from '@powersync/service-jpgwire';
 import { ReplicationMetric } from '@powersync/service-types';
 import * as crypto from 'crypto';
-import { describe, expect, test, it } from 'vitest';
+import { describe, expect, test } from 'vitest';
 import { describeWithStorage } from './util.js';
 import { WalStreamTestContext } from './wal_stream_utils.js';
 
@@ -325,7 +325,7 @@ bucket_definitions:
     }
   });
 
-  it.skip('custom types', async () => {
+  test('custom types', async () => {
     await using context = await WalStreamTestContext.open(factory);
 
     await context.updateSyncRules(`
@@ -346,6 +346,6 @@ config:
     await pool.query(`INSERT INTO test_data(id, description) VALUES ('t1', ROW(TRUE, 2)::composite)`);
 
     const data = await context.getBucketData('1#stream|0[]');
-    expect(data).toMatchObject([putOp('test_data', { id: 't1', description: 'test1' })]);
+    expect(data).toMatchObject([putOp('test_data', { id: 't1', description: '{"foo":1,"bar":2}' })]);
   });
 }