brianc#181 brianc#366 Add support for single row mode in Postgres 9.2+. This will enable single row mode only when the user wants to stream rows.

Author: rpedela

lib/native/index.js

@@ -130,18 +130,18 @@ Connection.prototype._pulseQueryQueue = function(initialConnection) {
   this._activeQuery = query;
   if(query.name) {
     if(this._namedQueries[query.name]) {
-      this._sendQueryPrepared(query.name, query.values||[]);
+      this._sendQueryPrepared(query.name, query.values||[], query.singleRowMode);
     } else {
       this._namedQuery = true;
       this._namedQueries[query.name] = true;
-      this._sendPrepare(query.name, query.text, (query.values||[]).length);
+      this._sendPrepare(query.name, query.text, (query.values||[]).length, query.singleRowMode);
     }
   } else if(query.values) {
     //call native function
-    this._sendQueryWithParams(query.text, query.values);
+    this._sendQueryWithParams(query.text, query.values, query.singleRowMode);
   } else {
     //call native function
-    this._sendQuery(query.text);
+    this._sendQuery(query.text, query.singleRowMode);
   }
 };
 

lib/native/query.js

@@ -20,6 +20,11 @@ var NativeQuery = function(config, values, callback) {
   this.text = c.text;
   this.values = c.values;
   this.callback = c.callback;
+  this.singleRowMode = false;
+
+  if(!this.callback) {
+    this.singleRowMode = true;
+  }
 
   this._result = new Result(config.rowMode);
   this._addedFields = false;

src/binding.cc

@@ -13,6 +13,10 @@
 #define ESCAPE_SUPPORTED
 #endif
 
+#if PG_VERSION_NUM >= 90200
+#define SINGLE_ROW_SUPPORTED
+#endif
+
 #define THROW(msg) return ThrowException(Exception::Error(String::New(msg)));
 
 using namespace v8;
@@ -204,7 +208,9 @@ class Connection : public ObjectWrap {
     }
 
     char* queryText = MallocCString(args[0]);
-    int result = self->Send(queryText);
+    bool singleRowMode = (bool)args[1]->Int32Value();
+
+    int result = self->Send(queryText, singleRowMode);
     free(queryText);
     if(result == 0) {
       lastErrorMessage = self->GetLastError();
@@ -234,7 +240,8 @@ class Connection : public ObjectWrap {
     String::Utf8Value queryName(args[0]);
     String::Utf8Value queryText(args[1]);
     int length = args[2]->Int32Value();
-    self->SendPrepare(*queryName, *queryText, length);
+    bool singleRowMode = (bool)args[3]->Int32Value();
+    self->SendPrepare(*queryName, *queryText, length, singleRowMode);
 
     return Undefined();
   }
@@ -274,12 +281,13 @@ class Connection : public ObjectWrap {
     }
 
     char* queryText = MallocCString(args[0]);
+    bool singleRowMode = (bool)args[2]->Int32Value();
 
     int result = 0;
     if(isPrepared) {
-      result = self->SendPreparedQuery(queryText, len, paramValues);
+      result = self->SendPreparedQuery(queryText, len, paramValues, singleRowMode);
     } else {
-      result = self->SendQueryParams(queryText, len, paramValues);
+      result = self->SendQueryParams(queryText, len, paramValues, singleRowMode);
     }
 
     free(queryText);
@@ -383,33 +391,53 @@ class Connection : public ObjectWrap {
   }
 #endif
 
-  int Send(const char *queryText)
+  void enableSingleRowMode(bool enable)
+  {
+#ifdef SINGLE_ROW_SUPPORTED
+    if(enable == true) {
+        int mode = PQsetSingleRowMode(connection_);
+        if(mode == 1) {
+            TRACE("PQsetSingleRowMode enabled")
+        } else {
+            TRACE("PQsetSingleRowMode disabled")
+        }
+    } else {
+        TRACE("PQsetSingleRowMode disabled")
+    }
+#endif
+  }
+
+  int Send(const char *queryText, bool singleRowMode)
   {
     TRACE("js::Send")
     int rv = PQsendQuery(connection_, queryText);
+    enableSingleRowMode(singleRowMode);
     StartWrite();
     return rv;
   }
 
-  int SendQueryParams(const char *command, const int nParams, const char * const *paramValues)
+  int SendQueryParams(const char *command, const int nParams, const char * const *paramValues, bool singleRowMode)
   {
     TRACE("js::SendQueryParams")
     int rv = PQsendQueryParams(connection_, command, nParams, NULL, paramValues, NULL, NULL, 0);
+    enableSingleRowMode(singleRowMode);
     StartWrite();
     return rv;
   }
 
-  int SendPrepare(const char *name, const char *command, const int nParams)
+  int SendPrepare(const char *name, const char *command, const int nParams, bool singleRowMode)
   {
     TRACE("js::SendPrepare")
     int rv = PQsendPrepare(connection_, name, command, nParams, NULL);
+    enableSingleRowMode(singleRowMode);
     StartWrite();
     return rv;
   }
 
-  int SendPreparedQuery(const char *name, int nParams, const char * const *paramValues)
+  int SendPreparedQuery(const char *name, int nParams, const char * const *paramValues, bool singleRowMode)
   {
     int rv = PQsendQueryPrepared(connection_, name, nParams, paramValues, NULL, NULL, 0);
+    enableSingleRowMode(singleRowMode);
     StartWrite();
     return rv;
   }
@@ -631,6 +659,7 @@ class Connection : public ObjectWrap {
     ExecStatusType status = PQresultStatus(result);
     switch(status) {
     case PGRES_TUPLES_OK:
+    case PGRES_SINGLE_TUPLE:
       {
         EmitRowDescription(result);
         HandleTuplesResult(result);