Node.js MongoDB Driver API

Source: lib/bulk/ordered.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 FindOperatorsOrdered instance (INTERNAL TYPE, do not instantiate directly)
  36.  * @class
  37.  * @return {FindOperatorsOrdered} a FindOperatorsOrdered instance.
  38.  */
  39. var FindOperatorsOrdered = function(self) {
  40.   this.s = self.s;
  41. };
  43. /**
  44.  * Add a single update document to the bulk operation
  45.  *
  46.  * @method
  47.  * @param {object} doc update operations
  48.  * @throws {MongoError}
  49.  * @return {OrderedBulkOperation}
  50.  */
  51. FindOperatorsOrdered.prototype.update = function(updateDocument) {
  52.   // Perform upsert
  53.   var upsert = typeof this.s.currentOp.upsert === 'boolean' ? this.s.currentOp.upsert : false;
  55.   // Establish the update command
  56.   var document = {
  57.     q: this.s.currentOp.selector,
  58.     u: updateDocument,
  59.     multi: true,
  60.     upsert: upsert
  61.   };
  63.   // Clear out current Op
  64.   this.s.currentOp = null;
  65.   // Add the update document to the list
  66.   return addToOperationsList(this, common.UPDATE, document);
  67. };
  69. /**
  70.  * Add a single update one document to the bulk operation
  71.  *
  72.  * @method
  73.  * @param {object} doc update operations
  74.  * @throws {MongoError}
  75.  * @return {OrderedBulkOperation}
  76.  */
  77. FindOperatorsOrdered.prototype.updateOne = function(updateDocument) {
  78.   // Perform upsert
  79.   var upsert = typeof this.s.currentOp.upsert === 'boolean' ? this.s.currentOp.upsert : false;
  81.   // Establish the update command
  82.   var document = {
  83.     q: this.s.currentOp.selector,
  84.     u: updateDocument,
  85.     multi: false,
  86.     upsert: upsert
  87.   };
  89.   // Clear out current Op
  90.   this.s.currentOp = null;
  91.   // Add the update document to the list
  92.   return addToOperationsList(this, common.UPDATE, document);
  93. };
  95. /**
  96.  * Add a replace one operation to the bulk operation
  97.  *
  98.  * @method
  99.  * @param {object} doc the new document to replace the existing one with
  100.  * @throws {MongoError}
  101.  * @return {OrderedBulkOperation}
  102.  */
  103. FindOperatorsOrdered.prototype.replaceOne = function(updateDocument) {
  104.   this.updateOne(updateDocument);
  105. };
  107. /**
  108.  * Upsert modifier for update bulk operation
  109.  *
  110.  * @method
  111.  * @throws {MongoError}
  112.  * @return {FindOperatorsOrdered}
  113.  */
  114. FindOperatorsOrdered.prototype.upsert = function() {
  115.   this.s.currentOp.upsert = true;
  116.   return this;
  117. };
  119. /**
  120.  * Add a remove one operation to the bulk operation
  121.  *
  122.  * @method
  123.  * @throws {MongoError}
  124.  * @return {OrderedBulkOperation}
  125.  */
  126. FindOperatorsOrdered.prototype.deleteOne = function() {
  127.   // Establish the update command
  128.   var document = {
  129.     q: this.s.currentOp.selector,
  130.     limit: 1
  131.   };
  133.   // Clear out current Op
  134.   this.s.currentOp = null;
  135.   // Add the remove document to the list
  136.   return addToOperationsList(this, common.REMOVE, document);
  137. };
  139. // Backward compatibility
  140. FindOperatorsOrdered.prototype.removeOne = FindOperatorsOrdered.prototype.deleteOne;
  142. /**
  143.  * Add a remove operation to the bulk operation
  144.  *
  145.  * @method
  146.  * @throws {MongoError}
  147.  * @return {OrderedBulkOperation}
  148.  */
  149. FindOperatorsOrdered.prototype.delete = function() {
  150.   // Establish the update command
  151.   var document = {
  152.     q: this.s.currentOp.selector,
  153.     limit: 0
  154.   };
  156.   // Clear out current Op
  157.   this.s.currentOp = null;
  158.   // Add the remove document to the list
  159.   return addToOperationsList(this, common.REMOVE, document);
  160. };
  162. // Backward compatibility
  163. FindOperatorsOrdered.prototype.remove = FindOperatorsOrdered.prototype.delete;
  165. // Add to internal list of documents
  166. var addToOperationsList = function(_self, docType, document) {
  167.   // Get the bsonSize
  168.   var bsonSize = bson.calculateObjectSize(document, {
  169.     checkKeys: false
  170.   });
  172.   // Throw error if the doc is bigger than the max BSON size
  173.   if (bsonSize >= _self.s.maxBatchSizeBytes) {
  174.     throw toError('document is larger than the maximum size ' + _self.s.maxBatchSizeBytes);
  175.   }
  177.   // Create a new batch object if we don't have a current one
  178.   if (_self.s.currentBatch == null) _self.s.currentBatch = new Batch(docType, _self.s.currentIndex);
  180.   // Check if we need to create a new batch
  181.   if (
  182.     _self.s.currentBatchSize + 1 >= _self.s.maxWriteBatchSize ||
  183.     _self.s.currentBatchSizeBytes + _self.s.currentBatchSizeBytes >= _self.s.maxBatchSizeBytes ||
  184.     _self.s.currentBatch.batchType !== docType
  185.   ) {
  186.     // Save the batch to the execution stack
  187.     _self.s.batches.push(_self.s.currentBatch);
  189.     // Create a new batch
  190.     _self.s.currentBatch = new Batch(docType, _self.s.currentIndex);
  192.     // Reset the current size trackers
  193.     _self.s.currentBatchSize = 0;
  194.     _self.s.currentBatchSizeBytes = 0;
  195.   } else {
  196.     // Update current batch size
  197.     _self.s.currentBatchSize = _self.s.currentBatchSize + 1;
  198.     _self.s.currentBatchSizeBytes = _self.s.currentBatchSizeBytes + bsonSize;
  199.   }
  201.   if (docType === common.INSERT) {
  202.     _self.s.bulkResult.insertedIds.push({ index: _self.s.currentIndex, _id: document._id });
  203.   }
  205.   // We have an array of documents
  206.   if (Array.isArray(document)) {
  207.     throw toError('operation passed in cannot be an Array');
  208.   } else {
  209.     _self.s.currentBatch.originalIndexes.push(_self.s.currentIndex);
  210.     _self.s.currentBatch.operations.push(document);
  211.     _self.s.currentBatchSizeBytes = _self.s.currentBatchSizeBytes + bsonSize;
  212.     _self.s.currentIndex = _self.s.currentIndex + 1;
  213.   }
  215.   // Return self
  216.   return _self;
  217. };
  219. /**
  220.  * Create a new OrderedBulkOperation instance (INTERNAL TYPE, do not instantiate directly)
  221.  * @class
  222.  * @property {number} length Get the number of operations in the bulk.
  223.  * @return {OrderedBulkOperation} a OrderedBulkOperation instance.
  224.  */
  225. function OrderedBulkOperation(topology, collection, options) {
  226.   options = options == null ? {} : options;
  227.   // TODO Bring from driver information in isMaster
  228.   var executed = false;
  230.   // Current item
  231.   var currentOp = null;
  233.   // Handle to the bson serializer, used to calculate running sizes
  234.   var bson = topology.bson;
  236.   // Namespace for the operation
  237.   var namespace = collection.collectionName;
  239.   // Set max byte size
  240.   var maxBatchSizeBytes =
  241.     topology.isMasterDoc && topology.isMasterDoc.maxBsonObjectSize
  242.       ? topology.isMasterDoc.maxBsonObjectSize
  243.       : 1024 * 1025 * 16;
  244.   var maxWriteBatchSize =
  245.     topology.isMasterDoc && topology.isMasterDoc.maxWriteBatchSize
  246.       ? topology.isMasterDoc.maxWriteBatchSize
  247.       : 1000;
  249.   // Get the write concern
  250.   var writeConcern = applyWriteConcern(shallowClone(options), { collection: collection }, options);
  251.   writeConcern = writeConcern.writeConcern;
  253.   // Get the promiseLibrary
  254.   var promiseLibrary = options.promiseLibrary || Promise;
  256.   // Final results
  257.   var bulkResult = {
  258.     ok: 1,
  259.     writeErrors: [],
  260.     writeConcernErrors: [],
  261.     insertedIds: [],
  262.     nInserted: 0,
  263.     nUpserted: 0,
  264.     nMatched: 0,
  265.     nModified: 0,
  266.     nRemoved: 0,
  267.     upserted: []
  268.   };
  270.   // Internal state
  271.   this.s = {
  272.     // Final result
  273.     bulkResult: bulkResult,
  274.     // Current batch state
  275.     currentBatch: null,
  276.     currentIndex: 0,
  277.     currentBatchSize: 0,
  278.     currentBatchSizeBytes: 0,
  279.     batches: [],
  280.     // Write concern
  281.     writeConcern: writeConcern,
  282.     // Max batch size options
  283.     maxBatchSizeBytes: maxBatchSizeBytes,
  284.     maxWriteBatchSize: maxWriteBatchSize,
  285.     // Namespace
  286.     namespace: namespace,
  287.     // BSON
  288.     bson: bson,
  289.     // Topology
  290.     topology: topology,
  291.     // Options
  292.     options: options,
  293.     // Current operation
  294.     currentOp: currentOp,
  295.     // Executed
  296.     executed: executed,
  297.     // Collection
  298.     collection: collection,
  299.     // Promise Library
  300.     promiseLibrary: promiseLibrary,
  301.     // Fundamental error
  302.     err: null,
  303.     // Bypass validation
  304.     bypassDocumentValidation:
  305.       typeof options.bypassDocumentValidation === 'boolean'
  306.         ? options.bypassDocumentValidation
  307.         : false,
  308.     // check keys
  309.     checkKeys: typeof options.checkKeys === 'boolean' ? options.checkKeys : true
  310.   };
  311. }
  313. OrderedBulkOperation.prototype.raw = function(op) {
  314.   var key = Object.keys(op)[0];
  316.   // Set up the force server object id
  317.   var forceServerObjectId =
  318.     typeof this.s.options.forceServerObjectId === 'boolean'
  319.       ? this.s.options.forceServerObjectId
  320.       : this.s.collection.s.db.options.forceServerObjectId;
  322.   // Update operations
  323.   if (
  324.     (op.updateOne && op.updateOne.q) ||
  325.     (op.updateMany && op.updateMany.q) ||
  326.     (op.replaceOne && op.replaceOne.q)
  327.   ) {
  328.     op[key].multi = op.updateOne || op.replaceOne ? false : true;
  329.     return addToOperationsList(this, common.UPDATE, op[key]);
  330.   }
  332.   // Crud spec update format
  333.   if (op.updateOne || op.updateMany || op.replaceOne) {
  334.     var multi = op.updateOne || op.replaceOne ? false : true;
  335.     var operation = { q: op[key].filter, u: op[key].update || op[key].replacement, multi: multi };
  336.     operation.upsert = op[key].upsert ? true : false;
  337.     if (op.collation) operation.collation = op.collation;
  338.     if (op[key].arrayFilters) operation.arrayFilters = op[key].arrayFilters;
  339.     return addToOperationsList(this, common.UPDATE, operation);
  340.   }
  342.   // Remove operations
  343.   if (
  344.     op.removeOne ||
  345.     op.removeMany ||
  346.     (op.deleteOne && op.deleteOne.q) ||
  347.     (op.deleteMany && op.deleteMany.q)
  348.   ) {
  349.     op[key].limit = op.removeOne ? 1 : 0;
  350.     return addToOperationsList(this, common.REMOVE, op[key]);
  351.   }
  353.   // Crud spec delete operations, less efficient
  354.   if (op.deleteOne || op.deleteMany) {
  355.     var limit = op.deleteOne ? 1 : 0;
  356.     operation = { q: op[key].filter, limit: limit };
  357.     if (op.collation) operation.collation = op.collation;
  358.     return addToOperationsList(this, common.REMOVE, operation);
  359.   }
  361.   // Insert operations
  362.   if (op.insertOne && op.insertOne.document == null) {
  363.     if (forceServerObjectId !== true && op.insertOne._id == null) op.insertOne._id = new ObjectID();
  364.     return addToOperationsList(this, common.INSERT, op.insertOne);
  365.   } else if (op.insertOne && op.insertOne.document) {
  366.     if (forceServerObjectId !== true && op.insertOne.document._id == null)
  367.       op.insertOne.document._id = new ObjectID();
  368.     return addToOperationsList(this, common.INSERT, op.insertOne.document);
  369.   }
  371.   if (op.insertMany) {
  372.     for (var i = 0; i < op.insertMany.length; i++) {
  373.       if (forceServerObjectId !== true && op.insertMany[i]._id == null)
  374.         op.insertMany[i]._id = new ObjectID();
  375.       addToOperationsList(this, common.INSERT, op.insertMany[i]);
  376.     }
  378.     return;
  379.   }
  381.   // No valid type of operation
  382.   throw toError(
  383.     'bulkWrite only supports insertOne, insertMany, updateOne, updateMany, removeOne, removeMany, deleteOne, deleteMany'
  384.   );
  385. };
  387. /**
  388.  * Add a single insert document to the bulk operation
  389.  *
  390.  * @param {object} doc the document to insert
  391.  * @throws {MongoError}
  392.  * @return {OrderedBulkOperation}
  393.  */
  394. OrderedBulkOperation.prototype.insert = function(document) {
  395.   if (this.s.collection.s.db.options.forceServerObjectId !== true && document._id == null)
  396.     document._id = new ObjectID();
  397.   return addToOperationsList(this, common.INSERT, document);
  398. };
  400. /**
  401.  * Initiate a find operation for an update/updateOne/remove/removeOne/replaceOne
  402.  *
  403.  * @method
  404.  * @param {object} selector The selector for the bulk operation.
  405.  * @throws {MongoError}
  406.  * @return {FindOperatorsOrdered}
  407.  */
  408. OrderedBulkOperation.prototype.find = function(selector) {
  409.   if (!selector) {
  410.     throw toError('Bulk find operation must specify a selector');
  411.   }
  413.   // Save a current selector
  414.   this.s.currentOp = {
  415.     selector: selector
  416.   };
  418.   return new FindOperatorsOrdered(this);
  419. };
  421. Object.defineProperty(OrderedBulkOperation.prototype, 'length', {
  422.   enumerable: true,
  423.   get: function() {
  424.     return this.s.currentIndex;
  425.   }
  426. });
  428. //
  429. // Execute next write command in a chain
  430. var executeCommands = function(self, options, callback) {
  431.   if (self.s.batches.length === 0) {
  432.     return handleCallback(callback, null, new BulkWriteResult(self.s.bulkResult));
  433.   }
  435.   // Ordered execution of the command
  436.   var batch = self.s.batches.shift();
  438.   var resultHandler = function(err, result) {
  439.     // Error is a driver related error not a bulk op error, terminate
  440.     if ((err && err.driver) || (err && err.message)) {
  441.       return handleCallback(callback, err);
  442.     }
  444.     // If we have and error
  445.     if (err) err.ok = 0;
  446.     // Merge the results together
  447.     var mergeResult = mergeBatchResults(true, batch, self.s.bulkResult, err, result);
  448.     const writeResult = new BulkWriteResult(self.s.bulkResult);
  449.     if (mergeResult != null) {
  450.       return handleCallback(callback, null, writeResult);
  451.     }
  453.     // If we are ordered and have errors and they are
  454.     // not all replication errors terminate the operation
  455.     if (self.s.bulkResult.writeErrors.length > 0) {
  456.       if (self.s.bulkResult.writeErrors.length === 1) {
  457.         return handleCallback(
  458.           callback,
  459.           new BulkWriteError(toError(self.s.bulkResult.writeErrors[0]), writeResult),
  460.           null
  461.         );
  462.       }
  464.       return handleCallback(
  465.         callback,
  466.         new BulkWriteError(
  467.           toError({
  468.             message: 'write operation failed',
  469.             code: self.s.bulkResult.writeErrors[0].code,
  470.             writeErrors: self.s.bulkResult.writeErrors
  471.           }),
  472.           writeResult
  473.         ),
  474.         null
  475.       );
  476.     } else if (writeResult.getWriteConcernError()) {
  477.       return handleCallback(
  478.         callback,
  479.         new BulkWriteError(toError(writeResult.getWriteConcernError()), writeResult),
  480.         null
  481.       );
  482.     }
  484.     // Execute the next command in line
  485.     executeCommands(self, options, callback);
  486.   };
  488.   var finalOptions = Object.assign({ ordered: true }, options);
  489.   if (self.s.writeConcern != null) {
  490.     finalOptions.writeConcern = self.s.writeConcern;
  491.   }
  493.   // Set an operationIf if provided
  494.   if (self.operationId) {
  495.     resultHandler.operationId = self.operationId;
  496.   }
  498.   // Serialize functions
  499.   if (self.s.options.serializeFunctions) {
  500.     finalOptions.serializeFunctions = true;
  501.   }
  503.   // Ignore undefined
  504.   if (self.s.options.ignoreUndefined) {
  505.     finalOptions.ignoreUndefined = true;
  506.   }
  508.   // Is the bypassDocumentValidation options specific
  509.   if (self.s.bypassDocumentValidation === true) {
  510.     finalOptions.bypassDocumentValidation = true;
  511.   }
  513.   // Is the checkKeys option disabled
  514.   if (self.s.checkKeys === false) {
  515.     finalOptions.checkKeys = false;
  516.   }
  518.   if (finalOptions.retryWrites) {
  519.     if (batch.batchType === common.UPDATE) {
  520.       finalOptions.retryWrites = finalOptions.retryWrites && !batch.operations.some(op => op.multi);
  521.     }
  523.     if (batch.batchType === common.REMOVE) {
  524.       finalOptions.retryWrites =
  525.         finalOptions.retryWrites && !batch.operations.some(op => op.limit === 0);
  526.     }
  527.   }
  529.   try {
  530.     if (batch.batchType === common.INSERT) {
  531.       self.s.topology.insert(
  532.         self.s.collection.namespace,
  533.         batch.operations,
  534.         finalOptions,
  535.         resultHandler
  536.       );
  537.     } else if (batch.batchType === common.UPDATE) {
  538.       self.s.topology.update(
  539.         self.s.collection.namespace,
  540.         batch.operations,
  541.         finalOptions,
  542.         resultHandler
  543.       );
  544.     } else if (batch.batchType === common.REMOVE) {
  545.       self.s.topology.remove(
  546.         self.s.collection.namespace,
  547.         batch.operations,
  548.         finalOptions,
  549.         resultHandler
  550.       );
  551.     }
  552.   } catch (err) {
  553.     // Force top level error
  554.     err.ok = 0;
  555.     // Merge top level error and return
  556.     handleCallback(callback, null, mergeBatchResults(false, batch, self.s.bulkResult, err, null));
  557.   }
  558. };
  560. /**
  561.  * The callback format for results
  562.  * @callback OrderedBulkOperation~resultCallback
  563.  * @param {MongoError} error An error instance representing the error during the execution.
  564.  * @param {BulkWriteResult} result The bulk write result.
  565.  */
  567. /**
  568.  * Execute the ordered bulk operation
  569.  *
  570.  * @method
  571.  * @param {object} [options=null] Optional settings.
  572.  * @param {(number|string)} [options.w=null] The write concern.
  573.  * @param {number} [options.wtimeout=null] The write concern timeout.
  574.  * @param {boolean} [options.j=false] Specify a journal write concern.
  575.  * @param {boolean} [options.fsync=false] Specify a file sync write concern.
  576.  * @param {OrderedBulkOperation~resultCallback} [callback] The result callback
  577.  * @throws {MongoError}
  578.  * @return {Promise} returns Promise if no callback passed
  579.  */
  580. OrderedBulkOperation.prototype.execute = function(_writeConcern, options, callback) {
  581.   if (typeof options === 'function') (callback = options), (options = {});
  582.   options = options || {};
  584.   if (this.s.executed) {
  585.     var executedError = toError('batch cannot be re-executed');
  586.     return typeof callback === 'function'
  587.       ? callback(executedError, null)
  588.       : this.s.promiseLibrary.reject(executedError);
  589.   }
  591.   if (typeof _writeConcern === 'function') {
  592.     callback = _writeConcern;
  593.   } else if (_writeConcern && typeof _writeConcern === 'object') {
  594.     this.s.writeConcern = _writeConcern;
  595.   }
  597.   // If we have current batch
  598.   if (this.s.currentBatch) this.s.batches.push(this.s.currentBatch);
  600.   // If we have no operations in the bulk raise an error
  601.   if (this.s.batches.length === 0) {
  602.     var emptyBatchError = toError('Invalid Operation, no operations specified');
  603.     return typeof callback === 'function'
  604.       ? callback(emptyBatchError, null)
  605.       : this.s.promiseLibrary.reject(emptyBatchError);
  606.   }
  608.   return executeOperation(this.s.topology, executeCommands, [this, options, callback]);
  609. };
  611. /**
  612.  * Returns an unordered batch object
  613.  * @ignore
  614.  */
  615. var initializeOrderedBulkOp = function(topology, collection, options) {
  616.   return new OrderedBulkOperation(topology, collection, options);
  617. };
  619. initializeOrderedBulkOp.OrderedBulkOperation = OrderedBulkOperation;
  620. module.exports = initializeOrderedBulkOp;
  621. module.exports.Bulk = OrderedBulkOperation;