node-drivers: node-postgres based driver for testing (#4151)

* node-drivers: implement node-postgres based test connector

To facilitate local evaluation and performance testing of the prototype.
It follows the same patterns as the planetscale and neon drivers, and
includes smoke tests.

Co-authored-by: jkomyno <schiabel@prisma.io>

* chore: fix "cargo fmt"

* chore(pg-js-connector): remove need for type casts

* chore(pg-js-connector): style formatting

* fix(pg-js-connector): fix smoke test

* chore(pg-js-connector): reproduced test setup

* chore(pg-js-connector): removed unused deps, set author as @tomhoule

---------

Co-authored-by: jkomyno <schiabel@prisma.io>
Co-authored-by: jkomyno <skiabo97@gmail.com>

psl/builtin-connectors/src/lib.rs

@@ -16,6 +16,7 @@ mod mssql_datamodel_connector;
 mod mysql_datamodel_connector;
 mod native_type_definition;
 mod neon;
+mod pg_js;
 mod planetscale;
 mod postgres_datamodel_connector;
 mod sqlite_datamodel_connector;
@@ -30,13 +31,15 @@ pub const MSSQL: &'static dyn Connector = &mssql_datamodel_connector::MsSqlDatam
 pub const MONGODB: &'static dyn Connector = &mongodb::MongoDbDatamodelConnector;
 pub static PLANETSCALE_SERVERLESS: &'static dyn Connector = &planetscale::PLANETSCALE_SERVERLESS;
 pub static NEON_SERVERLESS: &'static dyn Connector = &neon::NEON_SERVERLESS;
+pub static PG_JS: &'static dyn Connector = &pg_js::PG_JS;
 
 pub static BUILTIN_CONNECTORS: ConnectorRegistry = &[
     POSTGRES,
     MYSQL,
     SQLITE,
     MSSQL,
     COCKROACH,
+    PG_JS,
     MONGODB,
     PLANETSCALE_SERVERLESS,
     NEON_SERVERLESS,

psl/builtin-connectors/src/pg_js.rs

@@ -0,0 +1,11 @@
+use crate::postgres_datamodel_connector;
+use psl_core::{datamodel_connector::Flavour, js_connector::JsConnector};
+
+pub(crate) static PG_JS: JsConnector = JsConnector {
+    flavour: Flavour::Postgres,
+    canonical_connector: &postgres_datamodel_connector::PostgresDatamodelConnector,
+
+    provider_name: "@prisma/pg",
+    name: "node-postgres (pg) connector",
+    allowed_protocols: Some(&["postgres", "postgresql"]),
+};

query-engine/js-connectors/js/pg-js-connector/package.json

@@ -0,0 +1,27 @@
+{
+  "name": "@jkomyno/prisma-pg-js-connector",
+  "version": "0.0.1",
+  "description": "Prisma's JS Connector for \"pg\"",
+  "main": "dist/index.js",
+  "module": "dist/index.mjs",
+  "types": "dist/index.d.ts",
+  "scripts": {
+    "build": "tsup ./src/index.ts --format cjs,esm --dts",
+    "lint": "tsc -p ./tsconfig.build.json"
+  },
+  "files": [
+    "dist",
+    "README.md"
+  ],
+  "keywords": [],
+  "author": "Tom Houlé <houle@prisma.io>",
+  "license": "Apache-2.0",
+  "sideEffects": false,
+  "dependencies": {
+    "@jkomyno/prisma-js-connector-utils": "workspace:*",
+    "pg": "^8.11.2"
+  },
+  "devDependencies": {
+    "@types/pg": "^8.10.2"
+  }
+}

query-engine/js-connectors/js/pg-js-connector/src/conversion.ts

@@ -0,0 +1,51 @@
+import { ColumnTypeEnum, type ColumnType } from '@jkomyno/prisma-js-connector-utils'
+import { types } from 'pg'
+
+const NeonColumnType = types.builtins
+
+/**
+ * This is a simplification of quaint's value inference logic. Take a look at quaint's conversion.rs
+ * module to see how other attributes of the field packet such as the field length are used to infer
+ * the correct quaint::Value variant.
+ */
+export function fieldToColumnType(fieldTypeId: number): ColumnType {
+  switch (fieldTypeId) {
+    case NeonColumnType['INT2']:
+    case NeonColumnType['INT4']:
+      return ColumnTypeEnum.Int32
+    case NeonColumnType['INT8']:
+      return ColumnTypeEnum.Int64
+    case NeonColumnType['FLOAT4']:
+      return ColumnTypeEnum.Float
+    case NeonColumnType['FLOAT8']:
+      return ColumnTypeEnum.Double
+    case NeonColumnType['BOOL']:
+      return ColumnTypeEnum.Boolean
+    case NeonColumnType['DATE']:
+      return ColumnTypeEnum.Date
+    case NeonColumnType['TIME']:
+      return ColumnTypeEnum.Time
+    case NeonColumnType['TIMESTAMP']:
+      return ColumnTypeEnum.DateTime
+    case NeonColumnType['NUMERIC']:
+      return ColumnTypeEnum.Numeric
+    case NeonColumnType['BPCHAR']:
+      return ColumnTypeEnum.Char
+    case NeonColumnType['TEXT']:
+    case NeonColumnType['VARCHAR']:
+      return ColumnTypeEnum.Text
+    case NeonColumnType['JSONB']:
+      return ColumnTypeEnum.Json
+    default:
+      if (fieldTypeId >= 10000) {
+        // Postgres Custom Types
+        return ColumnTypeEnum.Enum
+      }
+      throw new Error(`Unsupported column type: ${fieldTypeId}`)
+  }
+}
+
+// return string instead of JavaScript Date object
+types.setTypeParser(NeonColumnType.DATE, date => date)
+types.setTypeParser(NeonColumnType.TIME, date => date)
+types.setTypeParser(NeonColumnType.TIMESTAMP, date => date)

query-engine/js-connectors/js/pg-js-connector/src/index.ts

@@ -0,0 +1,2 @@
+export { createPgConnector } from './pg'
+export type { PrismaPgConfig } from './pg'

query-engine/js-connectors/js/pg-js-connector/src/pg.ts

@@ -0,0 +1,124 @@
+import * as pg from 'pg'
+import { bindConnector, bindTransaction, Debug } from '@jkomyno/prisma-js-connector-utils'
+import type { Connector, ConnectorConfig, Query, Queryable, ResultSet, Transaction } from '@jkomyno/prisma-js-connector-utils'
+import { fieldToColumnType } from './conversion'
+
+const debug = Debug('prisma:js-connector:pg')
+
+export type PrismaPgConfig = ConnectorConfig
+
+type StdClient = pg.Pool
+type TransactionClient = pg.PoolClient
+
+class PgQueryable<ClientT extends StdClient | TransactionClient>
+  implements Queryable {
+  readonly flavour = 'postgres'
+
+  constructor(protected readonly client: ClientT) {
+  }
+
+  /**
+   * Execute a query given as SQL, interpolating the given parameters.
+   */
+  async queryRaw(query: Query): Promise<ResultSet> {
+    const tag = '[js::query_raw]'
+    debug(`${tag} %O`, query)
+
+    const { fields, rows: results } = await this.performIO(query)
+
+    const columns = fields.map((field) => field.name)
+    const resultSet: ResultSet = {
+      columnNames: columns,
+      columnTypes: fields.map((field) => fieldToColumnType(field.dataTypeID)),
+      rows: results.map((result) => columns.map((column) => result[column])),
+    }
+
+    return resultSet
+  }
+
+  /**
+   * Execute a query given as SQL, interpolating the given parameters and
+   * returning the number of affected rows.
+   * Note: Queryable expects a u64, but napi.rs only supports u32.
+   */
+  async executeRaw(query: Query): Promise<number> {
+    const tag = '[js::execute_raw]'
+    debug(`${tag} %O`, query)
+
+    const { rowCount } = await this.performIO(query)
+    return rowCount
+  }
+
+  /**
+   * Run a query against the database, returning the result set.
+   * Should the query fail due to a connection error, the connection is
+   * marked as unhealthy.
+   */
+  private async performIO(query: Query) {
+    const { sql, args: values } = query
+
+    return await this.client.query(sql, values)
+  }
+}
+
+class PgTransaction extends PgQueryable<TransactionClient>
+  implements Transaction {
+  constructor(client: pg.PoolClient) {
+    super(client)
+  }
+
+  async commit(): Promise<void> {
+    const tag = '[js::commit]'
+    debug(`${tag} committing transaction`)
+
+    try {
+      await this.client.query('COMMIT')
+    } finally {
+      this.client.release()
+    }
+  }
+
+  async rollback(): Promise<void> {
+    const tag = '[js::rollback]'
+    debug(`${tag} rolling back the transaction`)
+
+    try {
+      await this.client.query('ROLLBACK')
+    } finally {
+      this.client.release()
+    }
+  }
+}
+
+class PrismaPg extends PgQueryable<StdClient> implements Connector {
+  constructor(config: PrismaPgConfig) {
+    const { url: connectionString } = config
+    const client = new pg.Pool({
+      connectionString,
+      ssl: {
+        rejectUnauthorized: false,
+      },
+    })
+    super(client)
+  }
+
+  async startTransaction(isolationLevel?: string): Promise<Transaction> {
+    const connection = await this.client.connect()
+    await connection.query('BEGIN')
+
+    if (isolationLevel) {
+      await connection.query(
+        `SET TRANSACTION ISOLATION LEVEL ${isolationLevel}`,
+      )
+    }
+
+    return bindTransaction(new PgTransaction(connection))
+  }
+
+  async close() {}
+}
+
+export const createPgConnector = (config: PrismaPgConfig): Connector => {
+  const db = new PrismaPg(config)
+  return bindConnector(db)
+}