Added pipeline

Author: mat50013

packages/pg/lib/client.js

@@ -42,7 +42,6 @@ class Client extends EventEmitter {
     this._connected = false
     this._connectionError = false
     this._queryable = true
-
     this.connection =
       c.connection ||
       new Connection({
@@ -52,10 +51,14 @@ class Client extends EventEmitter {
         keepAliveInitialDelayMillis: c.keepAliveInitialDelayMillis || 0,
         encoding: this.connectionParameters.client_encoding || 'utf8',
       })
+    this.hasExecuted = false;
     this.queryQueue = []
     this.binary = c.binary || defaults.binary
+    this.usePipeline = false;
+    this.preparedQueries = [];
     this.processID = null
     this.secretKey = null
+    this.hasQueriesPrepared = false;
     this.ssl = this.connectionParameters.ssl || false
     // As with Password, make SSL->Key (the private key) non-enumerable.
     // It won't show up in stack traces
@@ -311,7 +314,9 @@ class Client extends EventEmitter {
     if (activeQuery) {
       activeQuery.handleReadyForQuery(this.connection)
     }
-    this._pulseQueryQueue()
+    if (this.usePipeline == false) {
+      this._pulseQueryQueue()
+    }
   }
 
   // if we receieve an error event or error message
@@ -478,11 +483,10 @@ class Client extends EventEmitter {
 
   _pulseQueryQueue() {
     if (this.readyForQuery === true) {
-      this.activeQuery = this.queryQueue.shift()
+      this.activeQuery = this.queryQueue.shift();
       if (this.activeQuery) {
         this.readyForQuery = false
         this.hasExecuted = true
-
         const queryError = this.activeQuery.submit(this.connection)
         if (queryError) {
           process.nextTick(() => {
@@ -498,6 +502,120 @@ class Client extends EventEmitter {
     }
   }
 
+  async queryPipeline(queries) {
+    console.log("Entered pipeline");
+    this.usePipeline = true;
+    let queriesComputed = [];
+    let resultList = [];
+    for (let elem of queries) {
+      let queryToInsert, result;
+      [queryToInsert, result] = this.queryForPipeline(elem.config, elem.values);
+      queriesComputed.push(queryToInsert);
+      resultList.push(result);
+    }
+    for (let i = 0; i < queriesComputed.length; i++) {
+      console.log(`Processing query ${i}`);
+      let current = queriesComputed[i];
+      this.activeQuery = current;
+      let next;
+      if (i + 1 < queriesComputed.length) {
+        next = queriesComputed[i + 1];
+      }
+      if (!current.isPreparedSomehow()) {
+        await current.queryPrepareNew(this.connection);
+      } else {
+        console.log("Already prepared");
+      }
+      let pList = [];
+      pList.push(current.submitNew(this.connection));
+      if (next) {
+        pList.push(next.queryPrepareNew(this.connection));
+      }
+      await Promise.all(pList);
+      console.log(`Done processing query ${i}`);
+    }
+    this.activeQuery = null;
+    this.emit('drain');
+    this.usePipeline = false;
+    return resultList;
+  }
+
+  queryForPipeline(config, values, callback) {
+    // can take in strings, config object or query object
+    var query
+    var result
+    var readTimeout
+    var readTimeoutTimer
+    var queryCallback
+
+    if (config === null || config === undefined) {
+      throw new TypeError('Client was passed a null or undefined query')
+    } else if (typeof config.submit === 'function') {
+      readTimeout = config.query_timeout || this.connectionParameters.query_timeout
+      result = query = config
+      if (typeof values === 'function') {
+        query.callback = query.callback || values
+      }
+    } else {
+      readTimeout = this.connectionParameters.query_timeout
+      query = new Query(config, values, callback)
+      if (!query.callback) {
+        result = new this._Promise((resolve, reject) => {
+          query.callback = (err, res) => (err ? reject(err) : resolve(res))
+        }).catch(err => {
+          // replace the stack trace that leads to `TCP.onStreamRead` with one that leads back to the
+          // application that created the query
+          Error.captureStackTrace(err);
+          throw err;
+        })
+      }
+    }
+
+    if (readTimeout) {
+      queryCallback = query.callback
+
+      readTimeoutTimer = setTimeout(() => {
+        var error = new Error('Query read timeout')
+        process.nextTick(() => {
+          query.handleError(error, this.connection)
+        })
+        queryCallback(error)
+        // we already returned an error,
+        // just do nothing if query completes
+        query.callback = () => { }
+      }, readTimeout)
+
+      query.callback = (err, res) => {
+        clearTimeout(readTimeoutTimer)
+        queryCallback(err, res)
+      }
+    }
+
+    if (this.binary && !query.binary) {
+      query.binary = true
+    }
+
+    if (query._result && !query._result._types) {
+      query._result._types = this._types
+    }
+
+    if (!this._queryable) {
+      process.nextTick(() => {
+        query.handleError(new Error('Client has encountered a connection error and is not queryable'), this.connection)
+      })
+      return result
+    }
+
+    if (this._ending) {
+      process.nextTick(() => {
+        query.handleError(new Error('Client was closed and is not queryable'), this.connection)
+      })
+      return result
+    }
+
+    return [query, result]
+  }
+
   query(config, values, callback) {
     // can take in strings, config object or query object
     var query
@@ -543,15 +661,16 @@ class Client extends EventEmitter {
 
         // we already returned an error,
         // just do nothing if query completes
-        query.callback = () => {}
+        query.callback = () => { }
 
         // Remove from queue
         var index = this.queryQueue.indexOf(query)
         if (index > -1) {
           this.queryQueue.splice(index, 1)
         }
-
-        this._pulseQueryQueue()
+        if (this.usePipeline == false) {
+          this._pulseQueryQueue()
+        }
       }, readTimeout)
 
       query.callback = (err, res) => {
@@ -582,8 +701,10 @@ class Client extends EventEmitter {
       return result
     }
 
-    this.queryQueue.push(query)
-    this._pulseQueryQueue()
+    this.queryQueue.push(query);
+    if (this.usePipeline == false) {
+      this._pulseQueryQueue();
+    }
     return result
   }
 

packages/pg/lib/connection.js

@@ -148,7 +148,15 @@ class Connection extends EventEmitter {
     }
     return this.stream.write(buffer)
   }
-
+  writeArray(array) {
+    if (!this.stream.writable) {
+      return false
+    }
+    for (let buffer_ins of array) {
+      this.stream.write(buffer_ins);
+    }
+    return true;
+  }
   query(text) {
     this._send(serialize.query(text))
   }

packages/pg/lib/query.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const { EventEmitter } = require('events')
-
+const { parse, serialize } = require('pg-protocol')
 const Result = require('./result')
 const utils = require('./utils')
 
@@ -21,6 +21,7 @@ class Query extends EventEmitter {
     this.portal = config.portal || ''
     this.callback = config.callback
     this._rowMode = config.rowMode
+    this.toBeWrittenInBuffer = [];
     if (process.domain && config.callback) {
       this.callback = process.domain.bind(config.callback)
     }
@@ -29,6 +30,7 @@ class Query extends EventEmitter {
     // potential for multiple results
     this._results = this._result
     this.isPreparedStatement = false
+    this.alreadyPrepared = false;
     this._canceledDueToError = false
     this._promise = null
   }
@@ -134,6 +136,11 @@ class Query extends EventEmitter {
     if (this._canceledDueToError) {
       return this.handleError(this._canceledDueToError, con)
     }
+    if (this.submitRes) {
+      this.submitRes();
+      this.submitP = null;
+      this.submitRes = null;
+   }
     if (this.callback) {
       try {
         this.callback(null, this._results)
@@ -163,7 +170,60 @@ class Query extends EventEmitter {
     } else {
       connection.query(this.text)
     }
-    return null
+    return null;
+  }
+
+  async queryPrepareNew(connection) {
+    this.isPreparedStatement = true
+    this.toBeWrittenInBuffer.push(await serialize.parse({
+      text: this.text,
+      name: this.name,
+      types: this.types,
+    }));
+    this.dataParsed = true;
+    try {
+      this.toBeWrittenInBuffer.push(serialize.bind({
+        portal: this.portal,
+        statement: this.name,
+        values: this.values,
+        binary: this.binary,
+        valueMapper: utils.prepareValue,
+      }));
+    } catch (err) {
+      this.handleError(err, connection)
+      return
+    }
+
+    this.toBeWrittenInBuffer.push(await serialize.describe({
+      type: 'P',
+      name: this.portal || '',
+    }));
+  }
+
+  isPreparedSomehow(connection) {
+    return this.hasBeenParsed(connection) || this.dataParsed || this.requiresPreparation() == false;
+  }
+
+  async submitNew(connection) {
+    let res;
+    let p = new Promise((resolve) => {
+      res = resolve;
+    })
+    this.submitP = p;
+    this.submitRes = res;
+    console.log("Submitting");
+    if (this.requiresPreparation()) {
+      console.log("Is of type require preparation");
+      connection.writeArray(this.toBeWrittenInBuffer);
+      this._getRows(connection, this.rows)
+    } else {
+      console.log("Does not require preparation");
+      connection.query(this.text);
+    }
+    await this.submitP;
+    console.log("Done submitting");
+    this.dataParsed = undefined;
+    this.toBeWrittenInBuffer = [];
   }
 
   hasBeenParsed(connection) {
@@ -196,7 +256,7 @@ class Query extends EventEmitter {
     this.isPreparedStatement = true
 
     // TODO refactor this poor encapsulation
-    if (!this.hasBeenParsed(connection)) {
+    if (!this.hasBeenParsed(connection) && this.alreadyPrepared == false) {
       connection.parse({
         text: this.text,
         name: this.name,

packages/pg/lib/utils.js

@@ -13,7 +13,8 @@ function escapeElement(elementRepresentation) {
 // a different array separator.
 function arrayString(val) {
   var result = '{'
-  for (var i = 0; i < val.length; i++) {
+  var len = val.length;
+  for (var i = 0; i < len; i++) {
     if (i > 0) {
       result = result + ','
     }
@@ -43,6 +44,12 @@ var prepareValue = function (val, seen) {
   if (val instanceof Buffer) {
     return val
   }
+  if (typeof val == "string") {
+    return val;
+  }
+  if (typeof val == "number") {
+    return val.toString();
+  }
   if (ArrayBuffer.isView(val)) {
     var buf = Buffer.from(val.buffer, val.byteOffset, val.byteLength)
     if (buf.length === val.byteLength) {
@@ -170,8 +177,8 @@ const escapeIdentifier = function (str) {
 const escapeLiteral = function (str) {
   var hasBackslash = false
   var escaped = "'"
-
-  for (var i = 0; i < str.length; i++) {
+  var len = str.length;
+  for (var i = 0; i < len; i++) {
     var c = str[i]
     if (c === "'") {
       escaped += c + c