Actually fetch types

Author: simolus3

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

@@ -700,6 +700,9 @@ WHERE  oid = $1::regclass`,
     // Drop conflicting tables. This includes for example renamed tables.
     await batch.drop(result.dropTables);
 
+    // Ensure we have a description for custom types referenced in the table.
+    await this.connections.types.fetchTypes(referencedTypeIds);
+
     // Snapshot if:
     // 1. Snapshot is requested (false for initial snapshot, since that process handles it elsewhere)
     // 2. Snapshot is not already done, AND:
@@ -710,9 +713,6 @@ 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();

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

@@ -590,4 +590,24 @@ function defineTests(factory: storage.TestStorageFactory) {
 
     expect(failures).toEqual([]);
   });
+
+  test('custom types', async () => {
+    await using context = await WalStreamTestContext.open(factory);
+
+    await context.updateSyncRules(`
+streams:
+  stream:
+    query: SELECT id, * 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 context.replicateSnapshot();
+  });
 }

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

@@ -324,4 +324,28 @@ bucket_definitions:
       // creating a new replication slot.
     }
   });
+
+  test('custom types', async () => {
+    await using context = await WalStreamTestContext.open(factory);
+
+    await context.updateSyncRules(`
+streams:
+  stream:
+    query: SELECT id, * FROM "test_data"
+
+config:
+  edition: 2
+`);
+
+    const { pool } = context;
+    await pool.query(`DROP TABLE IF EXISTS test_data`);
+    await pool.query(`CREATE TYPE composite AS (foo bool, bar int4);`);
+    await pool.query(`CREATE TABLE test_data(id text primary key, description composite);`);
+
+    await context.initializeReplication();
+    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' })]);
+  });
 }