Adding support for pg-native

.github/workflows/coverage.yml

@@ -39,7 +39,7 @@ jobs:
         run: npm run test:coverage
         env:
           CI: true
-          DB: "postgres mysql mysql2 mssql sqlite3"
+          DB: "postgres pgnative mysql mysql2 mssql sqlite3"
           KNEX_TEST_TIMEOUT: 60000
 
       - name: Stop Databases

.github/workflows/integration-tests.yml

@@ -19,7 +19,7 @@ jobs:
       fail-fast: false
       matrix:
         node-version: [16.x, 14.x, 12.x, 10.x]
-        database-type: [postgres, mysql, mssql, sqlite3]
+        database-type: [postgres, pgnative, mysql, mssql, sqlite3]
 
     steps:
       - name: Checkout Repository

lib/constants.js

@@ -12,6 +12,7 @@ const SUPPORTED_CLIENTS = Object.freeze(
     'mysql2',
     'oracledb',
     'postgres',
+    'pgnative',
     'redshift',
     'sqlite3',
   ].concat(Object.keys(CLIENT_ALIASES))
@@ -23,6 +24,7 @@ const DRIVER_NAMES = Object.freeze({
   MySQL2: 'mysql2',
   Oracle: 'oracledb',
   PostgreSQL: 'pg',
+  PgNative: 'pgnative',
   Redshift: 'pg-redshift',
   SQLite: 'sqlite3',
 });

lib/dialects/pgnative/index.js

@@ -0,0 +1,59 @@
+// PostgreSQL Native Driver (pg-native)
+// -------
+const Client_PG = require('../postgres');
+
+class Client_PgNative extends Client_PG {
+  constructor(...args) {
+    super(...args);
+    this.driverName = 'pgnative';
+    this.canCancelQuery = true;
+  }
+
+  _driver() {
+    return require('pg').native;
+  }
+
+  _stream(connection, obj, stream, options) {
+    if (!obj.sql) throw new Error('The query is empty');
+
+    const client = this;
+    return new Promise((resolver, rejecter) => {
+      stream.on('error', rejecter);
+      stream.on('end', resolver);
+
+      return client
+        ._query(connection, obj)
+        .then((obj) => obj.response)
+        .then(({ rows }) => rows.forEach((row) => stream.write(row)))
+        .catch(function (err) {
+          stream.emit('error', err);
+        })
+        .then(function () {
+          stream.end();
+        });
+    });
+  }
+
+  async cancelQuery(connectionToKill) {
+    try {
+      return await this._wrappedCancelQueryCall(null, connectionToKill);
+    } catch (err) {
+      this.logger.warn(`Connection Error: ${err}`);
+    }
+  }
+
+  _wrappedCancelQueryCall(emptyConnection, connectionToKill) {
+    return new Promise(function (resolve, reject) {
+      connectionToKill.native.cancel(function (err) {
+        if (err) {
+          reject(err);
+          return;
+        }
+
+        resolve(true);
+      });
+    });
+  }
+}
+
+module.exports = Client_PgNative;

lib/dialects/postgres/index.js

@@ -62,34 +62,40 @@ class Client_PG extends Client {
     return `"${value.replace(/"/g, '""')}"${arrayAccessor}`;
   }
 
+  _acquireOnlyConnection() {
+    const connection = new this.driver.Client(this.connectionSettings);
+
+    return connection.connect().then(() => connection);
+  }
+
   // Get a raw connection, called by the `pool` whenever a new
   // connection needs to be added to the pool.
   acquireRawConnection() {
     const client = this;
-    return new Promise(function (resolver, rejecter) {
-      const connection = new client.driver.Client(client.connectionSettings);
-      connection.connect(function (err, connection) {
-        if (err) {
-          return rejecter(err);
-        }
+
+    return this._acquireOnlyConnection()
+      .then(function (connection) {
         connection.on('error', (err) => {
           connection.__knex__disposed = err;
         });
+
         connection.on('end', (err) => {
           connection.__knex__disposed = err || 'Connection ended unexpectedly';
         });
+
         if (!client.version) {
           return client.checkVersion(connection).then(function (version) {
             client.version = version;
-            resolver(connection);
+            return connection;
           });
         }
-        resolver(connection);
+
+        return connection;
+      })
+      .then(function setSearchPath(connection) {
+        client.setSchemaSearchPath(connection);
+        return connection;
       });
-    }).then(function setSearchPath(connection) {
-      client.setSchemaSearchPath(connection);
-      return connection;
-    });
   }
 
   // Used to explicitly close a connection, called internally by the pool

package.json

@@ -29,13 +29,16 @@
     "test:oracledb": "cross-env DB=oracledb npm run test:db",
     "test:sqlite": "cross-env DB=sqlite3 npm run test:db",
     "test:postgres": "cross-env DB=postgres npm run test:db",
+    "test:pgnative": "cross-env DB=pgnative npm run test:db",
     "test:tape": "node test/tape/index.js | tap-spec",
     "test:cli": "cross-env KNEX_PATH=../knex.js KNEX=bin/cli.js jake -f test/jake/Jakefile",
     "db:start": "docker-compose -f scripts/docker-compose.yml up --build -d mysql oracledbxe postgres mssql && docker-compose -f scripts/docker-compose.yml up waitmssql waitmysql waitpostgres waitoracledbxe",
     "db:start:no-oracle": "docker-compose -f scripts/docker-compose.yml up --build -d mysql postgres mssql && docker-compose -f scripts/docker-compose.yml up waitmssql waitmysql waitpostgres",
     "db:stop": "docker-compose -f scripts/docker-compose.yml down",
     "db:start:postgres": "docker-compose -f scripts/docker-compose.yml up --build -d postgres && docker-compose -f scripts/docker-compose.yml up waitpostgres",
     "db:stop:postgres": "docker-compose -f scripts/docker-compose.yml down",
+    "db:start:pgnative": "docker-compose -f scripts/docker-compose.yml up --build -d pgnative && docker-compose -f scripts/docker-compose.yml up waitpgnative",
+    "db:stop:pgnative": "docker-compose -f scripts/docker-compose.yml down",
     "db:start:mysql": "docker-compose -f scripts/docker-compose.yml up --build -d mysql && docker-compose -f scripts/docker-compose.yml up waitmysql",
     "db:stop:mysql": "docker-compose -f scripts/docker-compose.yml down",
     "db:start:oracle": "docker-compose -f scripts/docker-compose.yml up --build -d oracledbxe && docker-compose -f scripts/docker-compose.yml up waitoracledbxe",
@@ -72,6 +75,9 @@
     "pg": {
       "optional": true
     },
+    "pg-native": {
+      "optional": true
+    },
     "sqlite3": {
       "optional": true
     }
@@ -105,6 +111,7 @@
     "nyc": "^15.1.0",
     "oracledb": "^5.2.0",
     "pg": "^8.7.1",
+    "pg-native": "^3.0.0",
     "pg-query-stream": "^4.2.1",
     "prettier": "2.3.2",
     "rimraf": "^3.0.2",
@@ -177,6 +184,7 @@
     "mysql": false,
     "mysql2": false,
     "pg": false,
+    "pg-native": false,
     "pg-query-stream": false,
     "oracle": false,
     "sqlite3": false,

scripts/docker-compose.yml

@@ -84,6 +84,27 @@ services:
       - -c
       - 'until /usr/local/bin/psql postgres://testuser:knextest@postgres/knex_test -c "SELECT 1"; do sleep 5; done'
 
+  pgnative:
+    image: postgres:alpine
+    # see https://www.postgresql.org/docs/current/non-durability.html
+    command: '-c full_page_writes=off -c fsync=off -c synchronous_commit=off'
+    ports:
+      - '25433:5432'
+    environment:
+      - POSTGRES_USER=testuser
+      - POSTGRES_PASSWORD=knextest
+      - POSTGRES_DB=knex_test
+  waitpgnative:
+    image: postgres:alpine
+    links:
+      - pgnative
+    depends_on:
+      - pgnative
+    entrypoint:
+      - bash
+      - -c
+      - 'until /usr/local/bin/psql postgres://testuser:knextest@pgnative/knex_test -c "SELECT 1"; do sleep 5; done'
+
   oracledbxe:
     image: quillbuilduser/oracle-18-xe
     container_name: oracledbxe_container

scripts/stress-test/docker-compose.yml

@@ -7,11 +7,13 @@ services:
       - "8474:8474"
       - "23306:23306"
       - "25432:25432"
+      - "25433:25433"
       - "21521:21521"
       - "21433:21433"
     links:
       - "mysql"
       - "postgresql"
+      - "pgnative"
       - "oracledbxe"
       - "mssql"
 
@@ -31,6 +33,14 @@ services:
       - POSTGRES_PASSWORD=postgresrootpassword
       - POSTGRES_USER=postgres
 
+  pgnative:
+    image: mdillon/postgis
+    ports:
+      - "35433:5432"
+    environment:
+      - POSTGRES_PASSWORD=postgresrootpassword
+      - POSTGRES_USER=postgres
+
   oracledbxe:
     image: wnameless/oracle-xe-11g
     ports:

scripts/stress-test/knex-stress-test.js

@@ -13,6 +13,13 @@ const pg = Knex({
   pool: { max: 50 },
 });
 
+const pgnative = Knex({
+  client: 'pgnative',
+  connection:
+    'postgres://postgres:postgresrootpassword@localhost:25433/postgres',
+  pool: { max: 50 },
+});
+
 const mysql2 = Knex({
   client: 'mysql2',
   connection:
@@ -91,10 +98,10 @@ setInterval(() => {
   lastCounters = _.cloneDeep(counters);
 }, 2000);
 
-async function killConnectionsPg() {
-  return pg.raw(`SELECT pg_terminate_backend(pg_stat_activity.pid) 
+async function killConnectionsPg(client) {
+  return client.raw(`SELECT pg_terminate_backend(pg_stat_activity.pid)
     FROM pg_stat_activity
-    WHERE pg_stat_activity.datname = 'postgres' 
+    WHERE pg_stat_activity.datname = 'postgres'
       AND pid <> pg_backend_pid()`);
 }
 
@@ -154,6 +161,7 @@ async function main() {
   // create TCP proxies for simulating bad connections etc.
   async function recreateProxies() {
     await recreateProxy('postgresql', 25432, 5432);
+    await recreateProxy('postgresql', 25433, 5433);
     await recreateProxy('mysql', 23306, 3306);
     await recreateProxy('oracledbxe', 21521, 1521);
     await recreateProxy('mssql', 21433, 1433);
@@ -164,6 +172,9 @@ async function main() {
   loopQueries('PSQL:', pg.raw('select 1'));
   loopQueries('PSQL TO:', pg.raw('select 1').timeout(20));
 
+  loopQueries('PGNATIVE:', pgnative.raw('select 1'));
+  loopQueries('PGNATIVE TO:', pgnative.raw('select 1').timeout(20));
+
   loopQueries('MYSQL:', mysql.raw('select 1'));
   loopQueries('MYSQL TO:', mysql.raw('select 1').timeout(20));
 
@@ -181,7 +192,8 @@ async function main() {
     await delay(20); // kill everything every quite often from server side
     try {
       await Promise.all([
-        killConnectionsPg(),
+        killConnectionsPg(pg),
+        killConnectionsPg(pgnative),
         killConnectionsMyslq(mysql),
         //        killConnectionsMyslq(mysql2),
         killConnectionsMssql(),