Node.js MongoDB Driver API

Source: lib/bulk/unordered.js

  1. 'use strict';
  3. const common = require('./common');
  4. const utils = require('../utils');
  5. const toError = require('../utils').toError;
  6. const handleCallback = require('../utils').handleCallback;
  7. const shallowClone = utils.shallowClone;
  8. const BulkWriteResult = common.BulkWriteResult;
  9. const ObjectID = require('mongodb-core').BSON.ObjectID;
  10. const BSON = require('mongodb-core').BSON;
  11. const Batch = common.Batch;
  12. const mergeBatchResults = common.mergeBatchResults;
  13. const executeOperation = utils.executeOperation;
  14. const BulkWriteError = require('./common').BulkWriteError;
  15. const applyWriteConcern = utils.applyWriteConcern;
  17. var bson = new BSON([
  18.   BSON.Binary,
  19.   BSON.Code,
  20.   BSON.DBRef,
  21.   BSON.Decimal128,
  22.   BSON.Double,
  23.   BSON.Int32,
  24.   BSON.Long,
  25.   BSON.Map,
  26.   BSON.MaxKey,
  27.   BSON.MinKey,
  28.   BSON.ObjectId,
  29.   BSON.BSONRegExp,
  30.   BSON.Symbol,
  31.   BSON.Timestamp
  32. ]);
  34. /**
  35.  * Create a FindOperatorsUnordered instance (INTERNAL TYPE, do not instantiate directly)
  36.  * @class
  37.  * @property {number} length Get the number of operations in the bulk.
  38.  * @return {FindOperatorsUnordered} a FindOperatorsUnordered instance.
  39.  */
  40. var FindOperatorsUnordered = function(self) {
  41.   this.s = self.s;
  42. };
  44. /**
  45.  * Add a single update document to the bulk operation
  46.  *
  47.  * @method
  48.  * @param {object} updateDocument update operations
  49.  * @throws {MongoError}
  50.  * @return {FindOperatorsUnordered}
  51.  */
  52. FindOperatorsUnordered.prototype.update = function(updateDocument) {
  53.   // Perform upsert
  54.   var upsert = typeof this.s.currentOp.upsert === 'boolean' ? this.s.currentOp.upsert : false;
  56.   // Establish the update command
  57.   var document = {
  58.     q: this.s.currentOp.selector,
  59.     u: updateDocument,
  60.     multi: true,
  61.     upsert: upsert
  62.   };
  64.   // Clear out current Op
  65.   this.s.currentOp = null;
  66.   // Add the update document to the list
  67.   return addToOperationsList(this, common.UPDATE, document);
  68. };
  70. /**
  71.  * Add a single update one document to the bulk operation
  72.  *
  73.  * @method
  74.  * @param {object} updateDocument update operations
  75.  * @throws {MongoError}
  76.  * @return {FindOperatorsUnordered}
  77.  */
  78. FindOperatorsUnordered.prototype.updateOne = function(updateDocument) {
  79.   // Perform upsert
  80.   var upsert = typeof this.s.currentOp.upsert === 'boolean' ? this.s.currentOp.upsert : false;
  82.   // Establish the update command
  83.   var document = {
  84.     q: this.s.currentOp.selector,
  85.     u: updateDocument,
  86.     multi: false,
  87.     upsert: upsert
  88.   };
  90.   // Clear out current Op
  91.   this.s.currentOp = null;
  92.   // Add the update document to the list
  93.   return addToOperationsList(this, common.UPDATE, document);
  94. };
  96. /**
  97.  * Add a replace one operation to the bulk operation
  98.  *
  99.  * @method
  100.  * @param {object} updateDocument the new document to replace the existing one with
  101.  * @throws {MongoError}
  102.  * @return {FindOperatorsUnordered}
  103.  */
  104. FindOperatorsUnordered.prototype.replaceOne = function(updateDocument) {
  105.   this.updateOne(updateDocument);
  106. };
  108. /**
  109.  * Upsert modifier for update bulk operation
  110.  *
  111.  * @method
  112.  * @throws {MongoError}
  113.  * @return {FindOperatorsUnordered}
  114.  */
  115. FindOperatorsUnordered.prototype.upsert = function() {
  116.   this.s.currentOp.upsert = true;
  117.   return this;
  118. };
  120. /**
  121.  * Add a remove one operation to the bulk operation
  122.  *
  123.  * @method
  124.  * @throws {MongoError}
  125.  * @return {FindOperatorsUnordered}
  126.  */
  127. FindOperatorsUnordered.prototype.removeOne = function() {
  128.   // Establish the update command
  129.   var document = {
  130.     q: this.s.currentOp.selector,
  131.     limit: 1
  132.   };
  134.   // Clear out current Op
  135.   this.s.currentOp = null;
  136.   // Add the remove document to the list
  137.   return addToOperationsList(this, common.REMOVE, document);
  138. };
  140. /**
  141.  * Add a remove operation to the bulk operation
  142.  *
  143.  * @method
  144.  * @throws {MongoError}
  145.  * @return {FindOperatorsUnordered}
  146.  */
  147. FindOperatorsUnordered.prototype.remove = function() {
  148.   // Establish the update command
  149.   var document = {
  150.     q: this.s.currentOp.selector,
  151.     limit: 0
  152.   };
  154.   // Clear out current Op
  155.   this.s.currentOp = null;
  156.   // Add the remove document to the list
  157.   return addToOperationsList(this, common.REMOVE, document);
  158. };
  160. //
  161. // Add to the operations list
  162. //
  163. var addToOperationsList = function(_self, docType, document) {
  164.   // Get the bsonSize
  165.   var bsonSize = bson.calculateObjectSize(document, {
  166.     checkKeys: false
  167.   });
  168.   // Throw error if the doc is bigger than the max BSON size
  169.   if (bsonSize >= _self.s.maxBatchSizeBytes)
  170.     throw toError('document is larger than the maximum size ' + _self.s.maxBatchSizeBytes);
  171.   // Holds the current batch
  172.   _self.s.currentBatch = null;
  173.   // Get the right type of batch
  174.   if (docType === common.INSERT) {
  175.     _self.s.currentBatch = _self.s.currentInsertBatch;
  176.   } else if (docType === common.UPDATE) {
  177.     _self.s.currentBatch = _self.s.currentUpdateBatch;
  178.   } else if (docType === common.REMOVE) {
  179.     _self.s.currentBatch = _self.s.currentRemoveBatch;
  180.   }
  182.   // Create a new batch object if we don't have a current one
  183.   if (_self.s.currentBatch == null) _self.s.currentBatch = new Batch(docType, _self.s.currentIndex);
  185.   // Check if we need to create a new batch
  186.   if (
  187.     _self.s.currentBatch.size + 1 >= _self.s.maxWriteBatchSize ||
  188.     _self.s.currentBatch.sizeBytes + bsonSize >= _self.s.maxBatchSizeBytes ||
  189.     _self.s.currentBatch.batchType !== docType
  190.   ) {
  191.     // Save the batch to the execution stack
  192.     _self.s.batches.push(_self.s.currentBatch);
  194.     // Create a new batch
  195.     _self.s.currentBatch = new Batch(docType, _self.s.currentIndex);
  196.   }
  198.   // We have an array of documents
  199.   if (Array.isArray(document)) {
  200.     throw toError('operation passed in cannot be an Array');
  201.   } else {
  202.     _self.s.currentBatch.operations.push(document);
  203.     _self.s.currentBatch.originalIndexes.push(_self.s.currentIndex);
  204.     _self.s.currentIndex = _self.s.currentIndex + 1;
  205.   }
  207.   // Save back the current Batch to the right type
  208.   if (docType === common.INSERT) {
  209.     _self.s.currentInsertBatch = _self.s.currentBatch;
  210.     _self.s.bulkResult.insertedIds.push({
  211.       index: _self.s.bulkResult.insertedIds.length,
  212.       _id: document._id
  213.     });
  214.   } else if (docType === common.UPDATE) {
  215.     _self.s.currentUpdateBatch = _self.s.currentBatch;
  216.   } else if (docType === common.REMOVE) {
  217.     _self.s.currentRemoveBatch = _self.s.currentBatch;
  218.   }
  220.   // Update current batch size
  221.   _self.s.currentBatch.size = _self.s.currentBatch.size + 1;
  222.   _self.s.currentBatch.sizeBytes = _self.s.currentBatch.sizeBytes + bsonSize;
  224.   // Return self
  225.   return _self;
  226. };
  228. /**
  229.  * Create a new UnorderedBulkOperation instance (INTERNAL TYPE, do not instantiate directly)
  230.  * @class
  231.  * @property {number} length Get the number of operations in the bulk.
  232.  * @return {UnorderedBulkOperation} a UnorderedBulkOperation instance.
  233.  */
  234. var UnorderedBulkOperation = function(topology, collection, options) {
  235.   options = options == null ? {} : options;
  237.   // Get the namesspace for the write operations
  238.   var namespace = collection.collectionName;
  239.   // Used to mark operation as executed
  240.   var executed = false;
  242.   // Current item
  243.   // var currentBatch = null;
  244.   var currentOp = null;
  246.   // Handle to the bson serializer, used to calculate running sizes
  247.   var bson = topology.bson;
  249.   // Set max byte size
  250.   var maxBatchSizeBytes =
  251.     topology.isMasterDoc && topology.isMasterDoc.maxBsonObjectSize
  252.       ? topology.isMasterDoc.maxBsonObjectSize
  253.       : 1024 * 1025 * 16;
  254.   var maxWriteBatchSize =
  255.     topology.isMasterDoc && topology.isMasterDoc.maxWriteBatchSize
  256.       ? topology.isMasterDoc.maxWriteBatchSize
  257.       : 1000;
  259.   // Get the write concern
  260.   var writeConcern = applyWriteConcern(shallowClone(options), { collection: collection }, options);
  261.   writeConcern = writeConcern.writeConcern;
  263.   // Get the promiseLibrary
  264.   var promiseLibrary = options.promiseLibrary || Promise;
  266.   // Final results
  267.   var bulkResult = {
  268.     ok: 1,
  269.     writeErrors: [],
  270.     writeConcernErrors: [],
  271.     insertedIds: [],
  272.     nInserted: 0,
  273.     nUpserted: 0,
  274.     nMatched: 0,
  275.     nModified: 0,
  276.     nRemoved: 0,
  277.     upserted: []
  278.   };
  280.   // Internal state
  281.   this.s = {
  282.     // Final result
  283.     bulkResult: bulkResult,
  284.     // Current batch state
  285.     currentInsertBatch: null,
  286.     currentUpdateBatch: null,
  287.     currentRemoveBatch: null,
  288.     currentBatch: null,
  289.     currentIndex: 0,
  290.     batches: [],
  291.     // Write concern
  292.     writeConcern: writeConcern,
  293.     // Max batch size options
  294.     maxBatchSizeBytes: maxBatchSizeBytes,
  295.     maxWriteBatchSize: maxWriteBatchSize,
  296.     // Namespace
  297.     namespace: namespace,
  298.     // BSON
  299.     bson: bson,
  300.     // Topology
  301.     topology: topology,
  302.     // Options
  303.     options: options,
  304.     // Current operation
  305.     currentOp: currentOp,
  306.     // Executed
  307.     executed: executed,
  308.     // Collection
  309.     collection: collection,
  310.     // Promise Library
  311.     promiseLibrary: promiseLibrary,
  312.     // Bypass validation
  313.     bypassDocumentValidation:
  314.       typeof options.bypassDocumentValidation === 'boolean'
  315.         ? options.bypassDocumentValidation
  316.         : false,
  317.     // check keys
  318.     checkKeys: typeof options.checkKeys === 'boolean' ? options.checkKeys : true
  319.   };
  320. };
  322. /**
  323.  * Add a single insert document to the bulk operation
  324.  *
  325.  * @param {object} document the document to insert
  326.  * @throws {MongoError}
  327.  * @return {UnorderedBulkOperation}
  328.  */
  329. UnorderedBulkOperation.prototype.insert = function(document) {
  330.   if (this.s.collection.s.db.options.forceServerObjectId !== true && document._id == null)
  331.     document._id = new ObjectID();
  332.   return addToOperationsList(this, common.INSERT, document);
  333. };
  335. /**
  336.  * Initiate a find operation for an update/updateOne/remove/removeOne/replaceOne
  337.  *
  338.  * @method
  339.  * @param {object} selector The selector for the bulk operation.
  340.  * @throws {MongoError}
  341.  * @return {FindOperatorsUnordered}
  342.  */
  343. UnorderedBulkOperation.prototype.find = function(selector) {
  344.   if (!selector) {
  345.     throw toError('Bulk find operation must specify a selector');
  346.   }
  348.   // Save a current selector
  349.   this.s.currentOp = {
  350.     selector: selector
  351.   };
  353.   return new FindOperatorsUnordered(this);
  354. };
  356. Object.defineProperty(UnorderedBulkOperation.prototype, 'length', {
  357.   enumerable: true,
  358.   get: function() {
  359.     return this.s.currentIndex;
  360.   }
  361. });
  363. UnorderedBulkOperation.prototype.raw = function(op) {
  364.   var key = Object.keys(op)[0];
  366.   // Set up the force server object id
  367.   var forceServerObjectId =
  368.     typeof this.s.options.forceServerObjectId === 'boolean'
  369.       ? this.s.options.forceServerObjectId
  370.       : this.s.collection.s.db.options.forceServerObjectId;
  372.   // Update operations
  373.   if (
  374.     (op.updateOne && op.updateOne.q) ||
  375.     (op.updateMany && op.updateMany.q) ||
  376.     (op.replaceOne && op.replaceOne.q)
  377.   ) {
  378.     op[key].multi = op.updateOne || op.replaceOne ? false : true;
  379.     return addToOperationsList(this, common.UPDATE, op[key]);
  380.   }
  382.   // Crud spec update format
  383.   if (op.updateOne || op.updateMany || op.replaceOne) {
  384.     var multi = op.updateOne || op.replaceOne ? false : true;
  385.     var operation = { q: op[key].filter, u: op[key].update || op[key].replacement, multi: multi };
  386.     if (op[key].upsert) operation.upsert = true;
  387.     if (op[key].arrayFilters) operation.arrayFilters = op[key].arrayFilters;
  388.     return addToOperationsList(this, common.UPDATE, operation);
  389.   }
  391.   // Remove operations
  392.   if (
  393.     op.removeOne ||
  394.     op.removeMany ||
  395.     (op.deleteOne && op.deleteOne.q) ||
  396.     (op.deleteMany && op.deleteMany.q)
  397.   ) {
  398.     op[key].limit = op.removeOne ? 1 : 0;
  399.     return addToOperationsList(this, common.REMOVE, op[key]);
  400.   }
  402.   // Crud spec delete operations, less efficient
  403.   if (op.deleteOne || op.deleteMany) {
  404.     var limit = op.deleteOne ? 1 : 0;
  405.     operation = { q: op[key].filter, limit: limit };
  406.     return addToOperationsList(this, common.REMOVE, operation);
  407.   }
  409.   // Insert operations
  410.   if (op.insertOne && op.insertOne.document == null) {
  411.     if (forceServerObjectId !== true && op.insertOne._id == null) op.insertOne._id = new ObjectID();
  412.     return addToOperationsList(this, common.INSERT, op.insertOne);
  413.   } else if (op.insertOne && op.insertOne.document) {
  414.     if (forceServerObjectId !== true && op.insertOne.document._id == null)
  415.       op.insertOne.document._id = new ObjectID();
  416.     return addToOperationsList(this, common.INSERT, op.insertOne.document);
  417.   }
  419.   if (op.insertMany) {
  420.     for (var i = 0; i < op.insertMany.length; i++) {
  421.       if (forceServerObjectId !== true && op.insertMany[i]._id == null)
  422.         op.insertMany[i]._id = new ObjectID();
  423.       addToOperationsList(this, common.INSERT, op.insertMany[i]);
  424.     }
  426.     return;
  427.   }
  429.   // No valid type of operation
  430.   throw toError(
  431.     'bulkWrite only supports insertOne, insertMany, updateOne, updateMany, removeOne, removeMany, deleteOne, deleteMany'
  432.   );
  433. };
  435. //
  436. // Execute the command
  437. var executeBatch = function(self, batch, options, callback) {
  438.   var finalOptions = Object.assign({ ordered: false }, options);
  439.   if (self.s.writeConcern != null) {
  440.     finalOptions.writeConcern = self.s.writeConcern;
  441.   }
  443.   var resultHandler = function(err, result) {
  444.     // Error is a driver related error not a bulk op error, terminate
  445.     if ((err && err.driver) || (err && err.message)) {
  446.       return handleCallback(callback, err);
  447.     }
  449.     // If we have and error
  450.     if (err) err.ok = 0;
  451.     handleCallback(callback, null, mergeBatchResults(false, batch, self.s.bulkResult, err, result));
  452.   };
  454.   // Set an operationIf if provided
  455.   if (self.operationId) {
  456.     resultHandler.operationId = self.operationId;
  457.   }
  459.   // Serialize functions
  460.   if (self.s.options.serializeFunctions) {
  461.     finalOptions.serializeFunctions = true;
  462.   }
  464.   // Ignore undefined
  465.   if (self.s.options.ignoreUndefined) {
  466.     finalOptions.ignoreUndefined = true;
  467.   }
  469.   // Is the bypassDocumentValidation options specific
  470.   if (self.s.bypassDocumentValidation === true) {
  471.     finalOptions.bypassDocumentValidation = true;
  472.   }
  474.   // Is the checkKeys option disabled
  475.   if (self.s.checkKeys === false) {
  476.     finalOptions.checkKeys = false;
  477.   }
  479.   if (finalOptions.retryWrites) {
  480.     if (batch.batchType === common.UPDATE) {
  481.       finalOptions.retryWrites = finalOptions.retryWrites && !batch.operations.some(op => op.multi);
  482.     }
  484.     if (batch.batchType === common.REMOVE) {
  485.       finalOptions.retryWrites =
  486.         finalOptions.retryWrites && !batch.operations.some(op => op.limit === 0);
  487.     }
  488.   }
  490.   try {
  491.     if (batch.batchType === common.INSERT) {
  492.       self.s.topology.insert(
  493.         self.s.collection.namespace,
  494.         batch.operations,
  495.         finalOptions,
  496.         resultHandler
  497.       );
  498.     } else if (batch.batchType === common.UPDATE) {
  499.       self.s.topology.update(
  500.         self.s.collection.namespace,
  501.         batch.operations,
  502.         finalOptions,
  503.         resultHandler
  504.       );
  505.     } else if (batch.batchType === common.REMOVE) {
  506.       self.s.topology.remove(
  507.         self.s.collection.namespace,
  508.         batch.operations,
  509.         finalOptions,
  510.         resultHandler
  511.       );
  512.     }
  513.   } catch (err) {
  514.     // Force top level error
  515.     err.ok = 0;
  516.     // Merge top level error and return
  517.     handleCallback(callback, null, mergeBatchResults(false, batch, self.s.bulkResult, err, null));
  518.   }
  519. };
  521. //
  522. // Execute all the commands
  523. var executeBatches = function(self, options, callback) {
  524.   var numberOfCommandsToExecute = self.s.batches.length;
  525.   // Execute over all the batches
  526.   for (var i = 0; i < self.s.batches.length; i++) {
  527.     executeBatch(self, self.s.batches[i], options, function(err) {
  528.       // Count down the number of commands left to execute
  529.       numberOfCommandsToExecute = numberOfCommandsToExecute - 1;
  531.       // Execute
  532.       if (numberOfCommandsToExecute === 0) {
  533.         // Driver level error
  534.         if (err) return handleCallback(callback, err);
  536.         const writeResult = new BulkWriteResult(self.s.bulkResult);
  537.         if (self.s.bulkResult.writeErrors.length > 0) {
  538.           if (self.s.bulkResult.writeErrors.length === 1) {
  539.             return handleCallback(
  540.               callback,
  541.               new BulkWriteError(toError(self.s.bulkResult.writeErrors[0]), writeResult),
  542.               null
  543.             );
  544.           }
  546.           return handleCallback(
  547.             callback,
  548.             new BulkWriteError(
  549.               toError({
  550.                 message: 'write operation failed',
  551.                 code: self.s.bulkResult.writeErrors[0].code,
  552.                 writeErrors: self.s.bulkResult.writeErrors
  553.               }),
  554.               writeResult
  555.             ),
  556.             null
  557.           );
  558.         } else if (writeResult.getWriteConcernError()) {
  559.           return handleCallback(
  560.             callback,
  561.             new BulkWriteError(toError(writeResult.getWriteConcernError()), writeResult),
  562.             null
  563.           );
  564.         }
  566.         return handleCallback(callback, null, writeResult);
  567.       }
  568.     });
  569.   }
  570. };
  572. /**
  573.  * The callback format for results
  574.  * @callback UnorderedBulkOperation~resultCallback
  575.  * @param {MongoError} error An error instance representing the error during the execution.
  576.  * @param {BulkWriteResult} result The bulk write result.
  577.  */
  579. /**
  580.  * Execute the ordered bulk operation
  581.  *
  582.  * @method
  583.  * @param {object} [options=null] Optional settings.
  584.  * @param {(number|string)} [options.w=null] The write concern.
  585.  * @param {number} [options.wtimeout=null] The write concern timeout.
  586.  * @param {boolean} [options.j=false] Specify a journal write concern.
  587.  * @param {boolean} [options.fsync=false] Specify a file sync write concern.
  588.  * @param {UnorderedBulkOperation~resultCallback} [callback] The result callback
  589.  * @throws {MongoError}
  590.  * @return {Promise} returns Promise if no callback passed
  591.  */
  592. UnorderedBulkOperation.prototype.execute = function(_writeConcern, options, callback) {
  593.   if (typeof options === 'function') (callback = options), (options = {});
  594.   options = options || {};
  596.   if (this.s.executed) {
  597.     var executedError = toError('batch cannot be re-executed');
  598.     return typeof callback === 'function'
  599.       ? callback(executedError, null)
  600.       : this.s.promiseLibrary.reject(executedError);
  601.   }
  603.   if (typeof _writeConcern === 'function') {
  604.     callback = _writeConcern;
  605.   } else if (_writeConcern && typeof _writeConcern === 'object') {
  606.     this.s.writeConcern = _writeConcern;
  607.   }
  609.   // If we have current batch
  610.   if (this.s.currentInsertBatch) this.s.batches.push(this.s.currentInsertBatch);
  611.   if (this.s.currentUpdateBatch) this.s.batches.push(this.s.currentUpdateBatch);
  612.   if (this.s.currentRemoveBatch) this.s.batches.push(this.s.currentRemoveBatch);
  614.   // If we have no operations in the bulk raise an error
  615.   if (this.s.batches.length === 0) {
  616.     var emptyBatchError = toError('Invalid Operation, no operations specified');
  617.     return typeof callback === 'function'
  618.       ? callback(emptyBatchError, null)
  619.       : this.s.promiseLibrary.reject(emptyBatchError);
  620.   }
  622.   return executeOperation(this.s.topology, executeBatches, [this, options, callback]);
  623. };
  625. /**
  626.  * Returns an unordered batch object
  627.  * @ignore
  628.  */
  629. var initializeUnorderedBulkOp = function(topology, collection, options) {
  630.   return new UnorderedBulkOperation(topology, collection, options);
  631. };
  633. initializeUnorderedBulkOp.UnorderedBulkOperation = UnorderedBulkOperation;
  634. module.exports = initializeUnorderedBulkOp;
  635. module.exports.Bulk = UnorderedBulkOperation;