Merge pull request #135 from Sage/prepare

Prepared statements to speed up large insert/update operations

Author: joeferner

.gitignore

@@ -5,3 +5,4 @@ node_modules
 *.swp
 .lock*
 .idea
+tests-settings-custom.*

README.md

@@ -187,12 +187,12 @@ Javascript code:
 
 ### Querying large tables
 
-To query large tables you should use a reader:
+To query large tables you should use a _reader_:
 
-* `connection.reader(sql, args)` creates a reader
-* `reader.nextRow(callback)` returns the next row through the callback
+* `reader = connection.reader(sql, args)`: creates a reader
+* `reader.nextRow(callback)`: returns the next row through the callback
 * `reader.nextRows(count, callback)` returns the next `count` rows through the callback. `count` is optional and `nextRows` uses the prefetch row count when `count` is omitted.
-* `connection.setPrefetchRowCount(count)` configures the prefetch row count for the connection. Prefetching can have a dramatic impact on performance but uses more memory. 
+* `connection.setPrefetchRowCount(count)`: configures the prefetch row count for the connection. Prefetching can have a dramatic impact on performance but uses more memory. 
 
 Example:
 
@@ -221,6 +221,38 @@ doRead(function(err) {
 });
 ```
 
+### Large inserts or updates
+
+To insert or update a large number of records you should use _prepared statements_ rather than individual `execute` calls on the connection object:
+
+* `statement = connection.prepare(sql)`: creates a prepared statement.
+* `statement.execute(args, callback)`: executes the prepared statement with the values in `args`. You can call this repeatedly on the same `statement`.
+
+Example:
+
+```javascript
+
+function doInsert(stmt, records, cb) {
+	if (records.length > 0) {
+		stmt.execute([records.shift()], function(err, count) {
+			if (err) return cb(err);
+			if (count !== 1) return cb(new Error("bad count: " + count));
+			// recurse with remaining records
+			doInsert(stmt, records, cb);
+		});
+	} else {
+		// we are done
+		return cb();
+	}
+}
+
+var statement = connection.prepare("INSERT INTO users (id, firstName, lastName) VALUES (:1, :2, :3)");
+doInsert(statement, users, function(err) {
+	if (err) throw err; // or log it
+	console.log("all records inserted");	
+});
+```
+
 # Limitations/Caveats
 
 * Currently no native support for connection pooling (forthcoming; use generic-pool for now.)

binding.gyp

@@ -6,6 +6,7 @@
                    "src/oracle_bindings.cpp", 
                    "src/executeBaton.cpp",
                    "src/reader.cpp",
+                   "src/statement.cpp",
                    "src/outParam.cpp" ],
       "conditions": [
         ["OS=='mac'", {

src/connection.cpp

@@ -3,6 +3,7 @@
 #include "executeBaton.h"
 #include "commitBaton.h"
 #include "rollbackBaton.h"
+#include "statement.h"
 #include "reader.h"
 #include "outParam.h"
 #include <vector>
@@ -23,6 +24,7 @@ void Connection::Init(Handle<Object> target) {
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "execute", Execute);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "executeSync", ExecuteSync);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "readerHandle", CreateReader);
+  NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "prepare", Prepare);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "close", Close);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "isConnected", IsConnected);
   NODE_SET_PROTOTYPE_METHOD(uni::Deref(constructorTemplate), "setAutoCommit", SetAutoCommit);
@@ -74,6 +76,23 @@ uni::CallbackType Connection::Execute(const uni::FunctionCallbackInfo& args) {
   UNI_RETURN(scope, args, Undefined());
 }
 
+uni::CallbackType Connection::Prepare(const uni::FunctionCallbackInfo& args) {
+  UNI_SCOPE(scope);
+  Connection* connection = ObjectWrap::Unwrap<Connection>(args.This());
+
+  REQ_STRING_ARG(0, sql);
+
+  String::Utf8Value sqlVal(sql);
+
+  StatementBaton* baton = new StatementBaton(connection, *sqlVal, NULL);
+
+  Local<Object> statementHandle(uni::Deref(Statement::constructorTemplate)->GetFunction()->NewInstance());
+  Statement* statement = ObjectWrap::Unwrap<Statement>(statementHandle);
+  statement->setBaton(baton);
+
+  UNI_RETURN(scope, args, statementHandle);
+}
+
 uni::CallbackType Connection::CreateReader(const uni::FunctionCallbackInfo& args) {
   UNI_SCOPE(scope);
   Connection* connection = ObjectWrap::Unwrap<Connection>(args.This());
@@ -413,25 +432,34 @@ void Connection::EIO_AfterRollback(uv_work_t* req, int status) {
   delete req;
 }
 
-void Connection::EIO_Execute(uv_work_t* req) {
-  ExecuteBaton* baton = static_cast<ExecuteBaton*>(req->data);
-
+oracle::occi::Statement* Connection::CreateStatement(ExecuteBaton* baton) {
   baton->rows = NULL;
   baton->error = NULL;
 
-  oracle::occi::Statement* stmt = NULL;
-  oracle::occi::ResultSet* rs = NULL;
+  if (! baton->connection->m_connection) {
+    baton->error = new std::string("Connection already closed");
+    return NULL;
+  }
   try {
-    if(! baton->connection->m_connection) {
-      baton->error = new std::string("Connection already closed");
-      return;
-    }
-    stmt = baton->connection->m_connection->createStatement(baton->sql);
+    oracle::occi::Statement* stmt = baton->connection->m_connection->createStatement(baton->sql);
     stmt->setAutoCommit(baton->connection->m_autoCommit);
     if (baton->connection->m_prefetchRowCount > 0) stmt->setPrefetchRowCount(baton->connection->m_prefetchRowCount);
-    int outputParam = SetValuesOnStatement(stmt, baton);
-    if (baton->error) goto cleanup;
+    return stmt;
+  } catch(oracle::occi::SQLException &ex) {
+    baton->error = new string(ex.getMessage());
+    return NULL;
+  }
+}
+
+void Connection::ExecuteStatement(ExecuteBaton* baton, oracle::occi::Statement* stmt) {
+  oracle::occi::ResultSet* rs = NULL;
+
+  int outputParam = SetValuesOnStatement(stmt, baton);
+  if (baton->error) goto cleanup;
+
+  if (!baton->outputs) baton->outputs = new std::vector<output_t*>();
 
+  try {
     int status = stmt->execute();
     if(status == oracle::occi::Statement::UPDATE_COUNT_AVAILABLE) {
       baton->updateCount = stmt->getUpdateCount();
@@ -508,12 +536,22 @@ void Connection::EIO_Execute(uv_work_t* req) {
     baton->error = new string("Unknown Error");
   }
 cleanup:
-  if(stmt && rs) {
+  if (rs) {
     stmt->closeResultSet(rs);
     rs = NULL;
   }
-  if(stmt) {
-    if(baton->connection->m_connection) {
+}
+
+void Connection::EIO_Execute(uv_work_t* req) {
+  ExecuteBaton* baton = static_cast<ExecuteBaton*>(req->data);
+
+  oracle::occi::Statement* stmt = CreateStatement(baton);
+  if (baton->error) return;
+
+  ExecuteStatement(baton, stmt);
+
+  if (stmt) {
+    if (baton->connection->m_connection) {
       baton->connection->m_connection->terminateStatement(stmt);
     }
     stmt = NULL;

src/connection.h

@@ -77,11 +77,14 @@ namespace uni {
 class Connection : public ObjectWrap {
   friend class Reader;
   friend class ReaderBaton;
+  friend class Statement;
+  friend class StatementBaton;
 public:
   static void Init(Handle<Object> target);
   static uni::CallbackType New(const uni::FunctionCallbackInfo& args);
   static uni::CallbackType Execute(const uni::FunctionCallbackInfo& args);
   static uni::CallbackType ExecuteSync(const uni::FunctionCallbackInfo& args);
+  static uni::CallbackType Prepare(const uni::FunctionCallbackInfo& args);
   static uni::CallbackType CreateReader(const uni::FunctionCallbackInfo& args);
   static uni::CallbackType Close(const uni::FunctionCallbackInfo& args);
   static uni::CallbackType IsConnected(const uni::FunctionCallbackInfo& args);
@@ -105,6 +108,10 @@ class Connection : public ObjectWrap {
   oracle::occi::Environment* getEnvironment() { return m_environment; }
 
 protected:
+  // shared with Statement
+  static oracle::occi::Statement* CreateStatement(ExecuteBaton* baton);
+  static void ExecuteStatement(ExecuteBaton* baton, oracle::occi::Statement* stmt);
+
   // shared with Reader
   oracle::occi::Connection* getConnection() { return m_connection; }
   bool getAutoCommit() { return m_autoCommit; }

src/executeBaton.cpp

@@ -13,7 +13,8 @@ ExecuteBaton::ExecuteBaton(Connection* connection, const char* sql, v8::Local<v8
   }
   this->outputs = new std::vector<output_t*>();
   this->error = NULL;
-  CopyValuesToBaton(this, values);
+  if (values) CopyValuesToBaton(this, values);
+  this->rows = NULL;
 }
 
 ExecuteBaton::~ExecuteBaton() {
@@ -24,6 +25,13 @@ ExecuteBaton::~ExecuteBaton() {
     delete col;
   }
 
+  ResetValues();
+  ResetRows();
+  ResetOutputs();
+  ResetError();
+}
+
+void ExecuteBaton::ResetValues() {
   for (std::vector<value_t*>::iterator iterator = values.begin(), end = values.end(); iterator != end; ++iterator) {
 
     value_t* val = *iterator;
@@ -40,22 +48,11 @@ ExecuteBaton::~ExecuteBaton() {
     }
     delete val;
   }
-
-  ResetRows();
-
-  if(outputs) {
-    for (std::vector<output_t*>::iterator iterator = outputs->begin(), end = outputs->end(); iterator != end; ++iterator) {
-      output_t* o = *iterator;
-      delete o;
-    }
-    delete outputs;
-  }
-
-  if(error) delete error;
+  values.clear();
 }
 
 void ExecuteBaton::ResetRows() {
-  if(rows) {
+  if (rows) {
     for (std::vector<row_t*>::iterator iterator = rows->begin(), end = rows->end(); iterator != end; ++iterator) {
       row_t* currentRow = *iterator;
       delete currentRow;
@@ -66,6 +63,24 @@ void ExecuteBaton::ResetRows() {
   }
 }
 
+void ExecuteBaton::ResetOutputs() {
+  if (outputs) {
+    for (std::vector<output_t*>::iterator iterator = outputs->begin(), end = outputs->end(); iterator != end; ++iterator) {
+      output_t* o = *iterator;
+      delete o;
+    }
+    delete outputs;
+    outputs = NULL;
+  }
+}
+
+void ExecuteBaton::ResetError() {
+  if (error) {
+    delete error;
+    error = NULL;
+  }
+}
+
 double CallDateMethod(v8::Local<v8::Date> date, const char* methodName) {
   Handle<Value> args[1]; // should be zero but on windows the compiler will not allow a zero length array
   Local<Value> result = Local<Function>::Cast(date->Get(String::New(methodName)))->Call(date, 0, args);

src/executeBaton.h

@@ -61,7 +61,10 @@ class ExecuteBaton {
 public:
   ExecuteBaton(Connection* connection, const char* sql, v8::Local<v8::Array>* values, v8::Handle<v8::Function>* callback);
   ~ExecuteBaton();
+  void ResetValues();
   void ResetRows();
+  void ResetOutputs();
+  void ResetError();
 
   Connection *connection;
   v8::Persistent<v8::Function> callback;
@@ -73,7 +76,6 @@ class ExecuteBaton {
   std::string* error;
   int updateCount;
 
-private:
   static void CopyValuesToBaton(ExecuteBaton* baton, v8::Local<v8::Array>* values);
 };
 

src/oracle_bindings.cpp

@@ -1,6 +1,7 @@
 
 #include "connection.h"
 #include "oracle_bindings.h"
+#include "statement.h"
 #include "reader.h"
 #include "outParam.h"
 
@@ -187,6 +188,7 @@ extern "C" {
   static void init(Handle<Object> target) {
     OracleClient::Init(target);
     Connection::Init(target);
+    Statement::Init(target);
     Reader::Init(target);
     OutParam::Init(target);
   }

src/readerBaton.h

@@ -3,11 +3,11 @@
 #define _reader_baton_h_
 
 #include "connection.h"
-#include "executeBaton.h"
+#include "statementBaton.h"
 
-class ReaderBaton : public ExecuteBaton {
+class ReaderBaton : public StatementBaton {
 public:
-  ReaderBaton(Connection* connection, const char* sql, v8::Local<v8::Array>* values) : ExecuteBaton(connection, sql, values, NULL) {
+  ReaderBaton(Connection* connection, const char* sql, v8::Local<v8::Array>* values) : StatementBaton(connection, sql, values) {
     stmt = NULL;
     rs = NULL;
     done = false;
@@ -18,23 +18,15 @@ class ReaderBaton : public ExecuteBaton {
   }
 
   void ResetStatement() {
-	  if(stmt && rs) {
+	  if (stmt && rs) {
       stmt->closeResultSet(rs);
       rs = NULL;
     }
-    if(stmt) {
-      if(connection->getConnection()) {
-         connection->getConnection()->terminateStatement(stmt);
-      }
-      stmt = NULL;
-    }
+    StatementBaton::ResetStatement();
   }
 
-  oracle::occi::Statement* stmt;
   oracle::occi::ResultSet* rs;
   int count;
-  bool done;
-  bool busy;
 };
 
 #endif