refacorted insert and insertMany to leverage bulkWrite to allow for more than 1000 doc inserts

Author: christkv

lib/aggregation_cursor.js

@@ -152,8 +152,8 @@ for(var name in CoreCursor.prototype) {
  * @return {AggregationCursor}
  */
 AggregationCursor.prototype.batchSize = function(value) {
-  if(this.s.state == AggregationCursor.CLOSED || this.isDead()) throw new MongoError("Cursor is closed");
-  if(typeof value != 'number') throw new MongoError("batchSize requires an integer");
+  if(this.s.state == AggregationCursor.CLOSED || this.isDead()) throw MongoError.create({message: "Cursor is closed", driver:true });
+  if(typeof value != 'number') throw MongoError.create({message: "batchSize requires an integer", drvier:true });
   if(this.s.cmd.cursor) this.s.cmd.cursor.batchSize = value;
   this.setCursorBatchSize(value);
   return this;

lib/bulk/common.js

@@ -18,7 +18,7 @@ var REMOVE = 3
 var writeConcern = function(target, col, options) {
   if(options.w != null || options.j != null || options.fsync != null) {
     target.writeConcern = options;
-  } else if(col.writeConcern.w != null || col.writeConcern.j != null || col.writeConcern.fsync != null) {      
+  } else if(col.writeConcern.w != null || col.writeConcern.j != null || col.writeConcern.fsync != null) {
     target.writeConcern = col.writeConcern;
   }
 
@@ -43,7 +43,7 @@ var defineReadOnlyProperty = function(self, name, value) {
  * correctly after command execution
  * @ignore
  */
-var Batch = function(batchType, originalZeroIndex) {  
+var Batch = function(batchType, originalZeroIndex) {
   this.originalZeroIndex = originalZeroIndex;
   this.currentIndex = 0;
   this.originalIndexes = [];
@@ -108,7 +108,7 @@ var BulkWriteResult = function(bulkResult) {
    * @return {object}
    */
   this.getUpsertedIdAt = function(index) {
-    return bulkResult.upserted[index]; 
+    return bulkResult.upserted[index];
   }
 
   /**
@@ -186,7 +186,7 @@ var BulkWriteResult = function(bulkResult) {
       for(var i = 0; i < bulkResult.writeConcernErrors.length; i++) {
         var err = bulkResult.writeConcernErrors[i];
         errmsg = errmsg + err.errmsg;
-        
+
         // TODO: Something better
         if(i == 0) errmsg = errmsg + " and ";
       }
@@ -289,7 +289,7 @@ var mergeBatchResults = function(ordered, batch, bulkResult, err, result) {
       , op: batch.operations[0]
     };
 
-    bulkResult.writeErrors.push(new WriteError(writeError));    
+    bulkResult.writeErrors.push(new WriteError(writeError));
     return;
   } else if(result.ok == 0 && bulkResult.ok == 0) {
     return;
@@ -337,7 +337,7 @@ var mergeBatchResults = function(ordered, batch, bulkResult, err, result) {
     var nModified = result.nModified;
     bulkResult.nUpserted = bulkResult.nUpserted + nUpserted;
     bulkResult.nMatched = bulkResult.nMatched + (result.n - nUpserted);
-    
+
     if(typeof nModified == 'number') {
       bulkResult.nModified = bulkResult.nModified + nModified;
     } else {
@@ -390,4 +390,4 @@ exports.MULTIPLE_ERROR = MULTIPLE_ERROR;
 exports.UNKNOWN_ERROR = UNKNOWN_ERROR;
 exports.INSERT = INSERT;
 exports.UPDATE = UPDATE;
-exports.REMOVE = REMOVE;
+exports.REMOVE = REMOVE;

lib/bulk/ordered.js

@@ -283,6 +283,8 @@ function OrderedBulkOperation(topology, collection, options) {
     , collection: collection
     // Promise Library
     , promiseLibrary: promiseLibrary
+		// Fundamental error
+		, err: null
   }
 }
 
@@ -391,6 +393,11 @@ var executeCommands = function(self, callback) {
   var batch = self.s.batches.shift();
 
   var resultHandler = function(err, result) {
+		// Error is a driver related error not a bulk op error, terminate
+		if(err && err.driver || err && err.message) {
+			return callback(err);
+		}
+
     // If we have and error
     if(err) err.ok = 0;
     // Merge the results together
@@ -419,6 +426,11 @@ var executeCommands = function(self, callback) {
 		resultHandler.operationId = self.operationId;
 	}
 
+	// Serialize functions
+	if(self.s.options.serializeFunctions) {
+		finalOptions.serializeFunctions = true
+	}
+
   try {
     if(batch.batchType == common.INSERT) {
       self.s.topology.insert(self.s.collection.namespace, batch.operations, finalOptions, resultHandler);
@@ -473,7 +485,9 @@ OrderedBulkOperation.prototype.execute = function(_writeConcern, callback) {
   }
 
   // Execute using callback
-  if(typeof callback == 'function') return executeCommands(this, callback);
+  if(typeof callback == 'function') {
+		return executeCommands(this, callback);
+	}
 
   // Return a Promise
   return new this.s.promiseLibrary(function(resolve, reject) {

lib/bulk/unordered.js

@@ -399,6 +399,11 @@ var executeBatch = function(self, batch, callback) {
   }
 
   var resultHandler = function(err, result) {
+		// Error is a driver related error not a bulk op error, terminate
+		if(err && err.driver || err && err.message) {
+			return callback(err);
+		}
+
     // If we have and error
     if(err) err.ok = 0;
     callback(null, mergeBatchResults(false, batch, self.s.bulkResult, err, result));
@@ -409,6 +414,11 @@ var executeBatch = function(self, batch, callback) {
 		resultHandler.operationId = self.operationId;
 	}
 
+	// Serialize functions
+	if(self.s.options.serializeFunctions) {
+		finalOptions.serializeFunctions = true
+	}
+
   try {
     if(batch.batchType == common.INSERT) {
       self.s.topology.insert(self.s.collection.namespace, batch.operations, finalOptions, resultHandler);
@@ -429,13 +439,20 @@ var executeBatch = function(self, batch, callback) {
 // Execute all the commands
 var executeBatches = function(self, callback) {
   var numberOfCommandsToExecute = self.s.batches.length;
+	var error = null;
   // Execute over all the batches
   for(var i = 0; i < self.s.batches.length; i++) {
     executeBatch(self, self.s.batches[i], function(err, result) {
+			// Driver layer error capture it
+			if(err) error = err;
+			// Count down the number of commands left to execute
       numberOfCommandsToExecute = numberOfCommandsToExecute - 1;
 
       // Execute
       if(numberOfCommandsToExecute == 0) {
+				// Driver level error
+				if(error) return callback(error);
+				// Treat write errors
         var error = self.s.bulkResult.writeErrors.length > 0 ? self.s.bulkResult.writeErrors[0] : null;
         callback(error, new BulkWriteResult(self.s.bulkResult));
       }

lib/collection.js

@@ -277,7 +277,7 @@ Collection.prototype.find = function() {
 
   // Ensure the query is an object
   if(selector != null && typeof selector != 'object') {
-    throw new MongoError("query selector must be an object");
+    throw MongoError.create({message: "query selector must be an object", driver:true });
   }
 
   // Build the find command
@@ -357,7 +357,7 @@ Collection.prototype.insertOne = function(doc, options, callback) {
   var self = this;
   if(typeof options == 'function') callback = options, options = {};
   options = options || {};
-  if(Array.isArray(doc)) return callback(new MongoError('doc parameter must be an object'));
+  if(Array.isArray(doc)) return callback(MongoError.create({message: 'doc parameter must be an object', driver:true }));
 
   // Execute using callback
   if(typeof callback == 'function') return insertOne(self, doc, options, callback);
@@ -384,6 +384,25 @@ var insertOne = function(self, doc, options, callback) {
   });
 }
 
+var mapInserManyResults = function(docs, r) {
+  var ids = r.getInsertedIds();
+  var keys = Object.keys(ids);
+  var finalIds = new Array(keys);
+
+  for(var i = 0; i < keys.length; i++) {
+    if(ids[keys[i]]._id) {
+      finalIds[ids[keys[i]].index] = ids[keys[i]]._id;
+    }
+  }
+
+  return {
+    result: {ok: 1, n: r.insertedCount},
+    ops: docs,
+    insertedCount: r.insertedCount,
+    insertedIds: finalIds
+  }
+}
+
 /**
  * Inserts an array of documents into MongoDB.
  * @method
@@ -400,33 +419,47 @@ var insertOne = function(self, doc, options, callback) {
 Collection.prototype.insertMany = function(docs, options, callback) {
   var self = this;
   if(typeof options == 'function') callback = options, options = {};
-  options = options || {};
-  if(!Array.isArray(docs)) return callback(new MongoError('docs parameter must be an array of documents'));
+  options = options || {ordered:true};
+  if(!Array.isArray(docs)) return callback(MongoError.create({message: 'docs parameter must be an array of documents', driver:true }));
+
+  // Get the write concern options
+  if(typeof options.checkKeys != 'boolean') {
+    options.checkKeys = true;
+  }
+
+  // If keep going set unordered
+  options['serializeFunctions'] = options['serializeFunctions'] || self.s.serializeFunctions;
+
+  // Do we have keepGoing legacy o
+  if(options.forceServerObjectId == null || options.forceServerObjectId == false) {
+    // Add _id if not specified
+    for(var i = 0; i < docs.length; i++) {
+      if(docs[i]._id == null) docs[i]._id = self.s.pkFactory.createPk();
+    }
+  }
+
+  // Generate the bulk write operations
+  var operations = [{
+    insertMany: docs
+  }];
 
   // Execute using callback
-  if(typeof callback == 'function') return insertMany(self, docs, options, callback);
+  if(typeof callback == 'function') return bulkWrite(self, operations, options, function(err, r) {
+    if(err) return callback(err);
+    if(r.hasWriteErrors()) return callback(r);
+    callback(null, mapInserManyResults(docs, r));
+  });
 
   // Return a Promise
   return new this.s.promiseLibrary(function(resolve, reject) {
-    insertMany(self, docs, options, function(err, r) {
+    bulkWrite(self, operations, options, function(err, r) {
       if(err) return reject(err);
-      resolve(r);
-    });
-  });
-}
+      if(r.hasWriteErrors()) {
+        return reject(r);
+      }
 
-var insertMany = function(self, docs, options, callback) {
-  insertDocuments(self, docs, options, function(err, r) {
-    if(callback == null) return;
-    if(err && callback) return callback(err);
-    if(r == null) return callback(null, {result: {ok:1}});
-    r.insertedCount = r.result.n;
-    var ids = [];
-    for(var i = 0; i < docs.length; i++) {
-      if(docs[i]._id) ids.push(docs[i]._id);
-    }
-    r.insertedIds = ids;
-    if(callback) callback(null, r);
+      resolve(mapInserManyResults(docs, r));
+    });
   });
 }
 
@@ -482,7 +515,9 @@ Collection.prototype.bulkWrite = function(operations, options, callback) {
   if(typeof options == 'function') callback = options, options = {};
   options = options || {ordered:true};
 
-  if(!Array.isArray(operations)) throw new MongoError("operations must be an array of documents");
+  if(!Array.isArray(operations)) {
+    throw MongoError.create({message: "operations must be an array of documents", driver:true });
+  }
 
   // Execute using callback
   if(typeof callback == 'function') return bulkWrite(self, operations, options, callback);
@@ -497,7 +532,7 @@ Collection.prototype.bulkWrite = function(operations, options, callback) {
 }
 
 var bulkWrite = function(self, operations, options, callback) {
-  var bulk = options.ordered == true || options.ordered == null ? self.initializeOrderedBulkOp() : self.initializeUnorderedBulkOp();
+  var bulk = options.ordered == true || options.ordered == null ? self.initializeOrderedBulkOp(options) : self.initializeUnorderedBulkOp(options);
 
   // for each op go through and add to the bulk
   for(var i = 0; i < operations.length; i++) {
@@ -510,6 +545,7 @@ var bulkWrite = function(self, operations, options, callback) {
 
   // Execute the bulk
   bulk.execute(writeCon, function(err, r) {
+    if(err) return callback(err);
     r.insertedCount = r.nInserted;
     r.matchedCount = r.nMatched;
     r.modifiedCount = r.nModified || 0;
@@ -518,6 +554,9 @@ var bulkWrite = function(self, operations, options, callback) {
     r.upsertedIds = {};
     r.insertedIds = {};
 
+    // Update the n
+    r.n = r.insertedCount;
+
     // Inserted documents
     var inserted = r.getInsertedIds();
     // Map inserted ids
@@ -601,18 +640,14 @@ var insertDocuments = function(self, docs, options, callback) {
 Collection.prototype.insert = function(docs, options, callback) {
   var self = this;
   if(typeof options == 'function') callback = options, options = {};
-  options = options || {};
+  options = options || {ordered:false};
+  docs = !Array.isArray(docs) ? [docs] : docs;
 
-  // Execute using callback
-  if(typeof callback == 'function') return insertDocuments(self, docs, options, callback);
+  if(options.keepGoing == true) {
+    options.ordered = false;
+  }
 
-  // Return a Promise
-  return new this.s.promiseLibrary(function(resolve, reject) {
-    insertDocuments(self, docs, options, function(err, r) {
-      if(err) return reject(err);
-      resolve(r);
-    });
-  });
+  return this.insertMany(docs, options, callback);
 }
 
 /**
@@ -1172,7 +1207,10 @@ Collection.prototype.options = function(callback) {
 var options = function(self, callback) {
   self.s.db.listCollections({name: self.s.name}).toArray(function(err, collections) {
     if(err) return handleCallback(callback, err);
-    if(collections.length == 0) return handleCallback(callback, new MongoError(f("collection %s not found", self.s.namespace)));
+    if(collections.length == 0) {
+      return handleCallback(callback, MongoError.create({message: f("collection %s not found", self.s.namespace), driver:true }));
+    }
+
     handleCallback(callback, err, collections[0].options || null);
   });
 }

lib/command_cursor.js

@@ -157,8 +157,8 @@ for(var i = 0; i < methodsToInherit.length; i++) {
  * @return {CommandCursor}
  */
 CommandCursor.prototype.batchSize = function(value) {
-  if(this.s.state == CommandCursor.CLOSED || this.isDead()) throw new MongoError("Cursor is closed");
-  if(typeof value != 'number') throw new MongoError("batchSize requires an integer");
+  if(this.s.state == CommandCursor.CLOSED || this.isDead()) throw MongoError.create({message: "Cursor is closed", driver:true});
+  if(typeof value != 'number') throw MongoError.create({message: "batchSize requires an integer", driver:true});
   if(this.s.cmd.cursor) this.s.cmd.cursor.batchSize = value;
   this.setCursorBatchSize(value);
   return this;