Report correct type for schema

Author: simolus3

modules/module-postgres/src/api/PostgresRouteAPIAdapter.ts

@@ -9,8 +9,11 @@ import { getDebugTableInfo } from '../replication/replication-utils.js';
 import { KEEPALIVE_STATEMENT, PUBLICATION_NAME } from '../replication/WalStream.js';
 import * as types from '../types/types.js';
 import { getApplicationName } from '../utils/application-name.js';
+import { PostgresTypeCache } from '../types/cache.js';
+import { CustomTypeRegistry, isKnownType } from '../types/registry.js';
 
 export class PostgresRouteAPIAdapter implements api.RouteAPI {
+  private typeCache: PostgresTypeCache;
   connectionTag: string;
   // TODO this should probably be configurable one day
   publicationName = PUBLICATION_NAME;
@@ -31,6 +34,7 @@ export class PostgresRouteAPIAdapter implements api.RouteAPI {
     connectionTag?: string,
     private config?: types.ResolvedConnectionConfig
   ) {
+    this.typeCache = new PostgresTypeCache(config?.typeRegistry ?? new CustomTypeRegistry(), pool);
     this.connectionTag = connectionTag ?? sync_rules.DEFAULT_TAG;
   }
 
@@ -297,6 +301,7 @@ LEFT JOIN (
   SELECT
     attrelid,
     attname,
+    atttypid,
     format_type(atttypid, atttypmod) as data_type,
     (SELECT typname FROM pg_catalog.pg_type WHERE oid = atttypid) as pg_type,
     attnum,
@@ -311,6 +316,7 @@ LEFT JOIN (
 )
 GROUP BY schemaname, tablename, quoted_name`
     );
+    await this.typeCache.fetchTypesForSchema();
     const rows = pgwire.pgwireRows(results);
 
     let schemas: Record<string, service_types.DatabaseSchema> = {};
@@ -332,9 +338,11 @@ GROUP BY schemaname, tablename, quoted_name`
         if (pg_type.startsWith('_')) {
           pg_type = `${pg_type.substring(1)}[]`;
         }
+
+        const knownType = this.typeCache.registry.lookupType(Number(column.atttypid));
         table.columns.push({
           name: column.attname,
-          sqlite_type: sync_rules.expressionTypeFromPostgresType(pg_type).typeFlags,
+          sqlite_type: sync_rules.ExpressionType.fromTypeText(knownType.sqliteType()).typeFlags,
           type: column.data_type,
           internal_type: column.data_type,
           pg_type: pg_type

modules/module-postgres/src/module/PostgresModule.ts

@@ -69,7 +69,8 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
   private resolveConfig(config: types.PostgresConnectionConfig): types.ResolvedConnectionConfig {
     return {
       ...config,
-      ...types.normalizeConnectionConfig(config)
+      ...types.normalizeConnectionConfig(config),
+      typeRegistry: this.customTypes
     };
   }
 

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

@@ -31,7 +31,7 @@ export class PgManager {
   ) {
     // The pool is lazy - no connections are opened until a query is performed.
     this.pool = pgwire.connectPgWirePool(this.options, poolOptions);
-    this.types = new PostgresTypeCache(poolOptions.registry, this.pool, () => this.getServerVersion());
+    this.types = new PostgresTypeCache(poolOptions.registry, this.pool);
   }
 
   public get connectionTag() {

modules/module-postgres/src/types/cache.ts

@@ -2,6 +2,7 @@ import { DatabaseInputRow, SqliteInputRow, toSyncRulesRow } from '@powersync/ser
 import * as pgwire from '@powersync/service-jpgwire';
 import { CustomTypeRegistry } from './registry.js';
 import semver from 'semver';
+import { getServerVersion } from '../utils/postgres_version.js';
 
 /**
  * A cache of custom types for which information can be crawled from the source database.
@@ -11,15 +12,14 @@ export class PostgresTypeCache {
 
   constructor(
     readonly registry: CustomTypeRegistry,
-    private readonly pool: pgwire.PgClient,
-    private readonly getVersion: () => Promise<semver.SemVer | null>
+    private readonly pool: pgwire.PgClient
   ) {
     this.registry = new CustomTypeRegistry();
   }
 
   private async fetchVersion(): Promise<semver.SemVer> {
     if (this.cachedVersion == null) {
-      this.cachedVersion = (await this.getVersion()) ?? semver.parse('0.0.1');
+      this.cachedVersion = (await getServerVersion(this.pool)) ?? semver.parse('0.0.1');
     }
 
     return this.cachedVersion!;
@@ -138,9 +138,9 @@ WHERE t.oid = ANY($1)
   }
 
   /**
-   * Used for testing - fetches all custom types referenced by any column in the schema.
+   * Used for testing - fetches all custom types referenced by any column in the database.
    */
-  public async fetchTypesForSchema(schema: string = 'public') {
+  public async fetchTypesForSchema() {
     const sql = `
 SELECT DISTINCT a.atttypid AS type_oid
 FROM pg_attribute a
@@ -150,10 +150,10 @@ JOIN pg_type t ON t.oid = a.atttypid
 JOIN pg_namespace tn ON tn.oid = t.typnamespace
 WHERE a.attnum > 0
   AND NOT a.attisdropped
-  AND cn.nspname = $1
+  AND cn.nspname not in ('information_schema', 'pg_catalog', 'pg_toast')
     `;
 
-    const query = await this.pool.query({ statement: sql, params: [{ type: 'varchar', value: schema }] });
+    const query = await this.pool.query({ statement: sql });
     let ids: number[] = [];
     for (const row of pgwire.pgwireRows(query)) {
       ids.push(Number(row.type_oid));

modules/module-postgres/src/types/registry.ts

@@ -276,3 +276,7 @@ export class CustomTypeRegistry {
     }
   }
 }
+
+export function isKnownType(type: MaybeKnownType): type is KnownType {
+  return type.type != 'unknown';
+}

modules/module-postgres/src/types/types.ts

@@ -1,6 +1,7 @@
 import * as lib_postgres from '@powersync/lib-service-postgres';
 import * as service_types from '@powersync/service-types';
 import * as t from 'ts-codec';
+import { CustomTypeRegistry } from './registry.js';
 
 // Maintain backwards compatibility by exporting these
 export const validatePort = lib_postgres.validatePort;
@@ -24,7 +25,10 @@ export type PostgresConnectionConfig = t.Decoded<typeof PostgresConnectionConfig
 /**
  * Resolved version of {@link PostgresConnectionConfig}
  */
-export type ResolvedConnectionConfig = PostgresConnectionConfig & NormalizedPostgresConnectionConfig;
+export type ResolvedConnectionConfig = PostgresConnectionConfig &
+  NormalizedPostgresConnectionConfig & {
+    typeRegistry: CustomTypeRegistry;
+  };
 
 export function isPostgresConfig(
   config: service_types.configFile.DataSourceConfig

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

@@ -11,7 +11,7 @@ import { describe, expect, test } from 'vitest';
 import { clearTestDb, connectPgPool, connectPgWire, TEST_URI } from './util.js';
 import { WalStream } from '@module/replication/WalStream.js';
 import { PostgresTypeCache } from '@module/types/cache.js';
-import { getServerVersion } from '@module/utils/postgres_version.js';
+import { CustomTypeRegistry } from '@module/types/registry.js';
 
 describe('pg data types', () => {
   async function setupTable(db: pgwire.PgClient) {
@@ -551,7 +551,7 @@ INSERT INTO test_data(id, time, timestamp, timestamptz) VALUES (1, '17:42:01.12'
   test('test replication - multiranges', async () => {
     const db = await connectPgPool();
 
-    if (!(await new PostgresTypeCache(db, () => getServerVersion(db)).supportsMultiRanges())) {
+    if (!(await new PostgresTypeCache(new CustomTypeRegistry(), db).supportsMultiRanges())) {
       // This test requires Postgres 14 or later.
       return;
     }
@@ -620,7 +620,7 @@ INSERT INTO test_data(id, time, timestamp, timestamptz) VALUES (1, '17:42:01.12'
  * Return all the inserts from the first transaction in the replication stream.
  */
 async function getReplicationTx(db: pgwire.PgClient, replicationStream: pgwire.ReplicationStream) {
-  const typeCache = new PostgresTypeCache(db, () => getServerVersion(db));
+  const typeCache = new PostgresTypeCache(new CustomTypeRegistry(), db);
   await typeCache.fetchTypesForSchema();
 
   let transformed: SqliteInputRow[] = [];

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

@@ -0,0 +1,60 @@
+import { describe, expect, test } from 'vitest';
+import { clearTestDb, connectPgPool } from './util.js';
+import { PostgresRouteAPIAdapter } from '@module/api/PostgresRouteAPIAdapter.js';
+import { TYPE_INTEGER, TYPE_REAL, TYPE_TEXT } from '@powersync/service-sync-rules';
+
+describe('PostgresRouteAPIAdapter tests', () => {
+  test('infers connection schema', async () => {
+    const db = await connectPgPool();
+    try {
+      await clearTestDb(db);
+      const api = new PostgresRouteAPIAdapter(db);
+
+      await db.query(`CREATE DOMAIN rating_value AS FLOAT CHECK (VALUE BETWEEN 0 AND 5)`);
+      await db.query(`
+        CREATE TABLE test_users (
+            id TEXT NOT NULL PRIMARY KEY,
+            is_admin BOOLEAN,
+            rating RATING_VALUE
+        );
+      `);
+
+      const schema = await api.getConnectionSchema();
+      expect(schema).toStrictEqual([
+        {
+          name: 'public',
+          tables: [
+            {
+              name: 'test_users',
+              columns: [
+                {
+                  internal_type: 'text',
+                  name: 'id',
+                  pg_type: 'text',
+                  sqlite_type: TYPE_TEXT,
+                  type: 'text'
+                },
+                {
+                  internal_type: 'boolean',
+                  name: 'is_admin',
+                  pg_type: 'bool',
+                  sqlite_type: TYPE_INTEGER,
+                  type: 'boolean'
+                },
+                {
+                  internal_type: 'rating_value',
+                  name: 'rating',
+                  pg_type: 'rating_value',
+                  sqlite_type: TYPE_REAL,
+                  type: 'rating_value'
+                }
+              ]
+            }
+          ]
+        }
+      ]);
+    } finally {
+      await db.end();
+    }
+  });
+});

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

@@ -19,6 +19,7 @@ import { METRICS_HELPER, test_utils } from '@powersync/service-core-tests';
 import * as mongo_storage from '@powersync/service-module-mongodb-storage';
 import * as postgres_storage from '@powersync/service-module-postgres-storage';
 import * as timers from 'node:timers/promises';
+import { CustomTypeRegistry } from '@module/types/registry.js';
 
 describe.skipIf(!(env.CI || env.SLOW_TESTS))('slow tests', function () {
   describeWithStorage({ timeout: 120_000 }, function (factory) {
@@ -68,7 +69,7 @@ function defineSlowTests(factory: storage.TestStorageFactory) {
   });
 
   async function testRepeatedReplication(testOptions: { compact: boolean; maxBatchSize: number; numBatches: number }) {
-    const connections = new PgManager(TEST_CONNECTION_OPTIONS, {});
+    const connections = new PgManager(TEST_CONNECTION_OPTIONS, { registry: new CustomTypeRegistry() });
     const replicationConnection = await connections.replicationConnection();
     const pool = connections.pool;
     await clearTestDb(pool);
@@ -329,7 +330,7 @@ bucket_definitions:
       await pool.query(`SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE active = FALSE`);
       i += 1;
 
-      const connections = new PgManager(TEST_CONNECTION_OPTIONS, {});
+      const connections = new PgManager(TEST_CONNECTION_OPTIONS, { registry: new CustomTypeRegistry() });
       const replicationConnection = await connections.replicationConnection();
 
       abortController = new AbortController();

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

@@ -12,6 +12,7 @@ import {
 import { METRICS_HELPER, test_utils } from '@powersync/service-core-tests';
 import * as pgwire from '@powersync/service-jpgwire';
 import { clearTestDb, getClientCheckpoint, TEST_CONNECTION_OPTIONS } from './util.js';
+import { CustomTypeRegistry } from '@module/types/registry.js';
 
 export class WalStreamTestContext implements AsyncDisposable {
   private _walStream?: WalStream;
@@ -32,7 +33,7 @@ export class WalStreamTestContext implements AsyncDisposable {
     options?: { doNotClear?: boolean; walStreamOptions?: Partial<WalStreamOptions> }
   ) {
     const f = await factory({ doNotClear: options?.doNotClear });
-    const connectionManager = new PgManager(TEST_CONNECTION_OPTIONS, {});
+    const connectionManager = new PgManager(TEST_CONNECTION_OPTIONS, { registry: new CustomTypeRegistry() });
 
     if (!options?.doNotClear) {
       await clearTestDb(connectionManager.pool);