perf(NODE-7660): serialize bulk write documents a single time

src/bulk/common.ts

@@ -157,6 +157,7 @@ export class Batch<T = Document> {
   originalIndexes: number[];
   batchType: BatchType;
   operations: T[];
+  serializedOperations: Uint8Array[];
   size: number;
   sizeBytes: number;
 
@@ -166,6 +167,7 @@ export class Batch<T = Document> {
     this.originalIndexes = [];
     this.batchType = batchType;
     this.operations = [];
+    this.serializedOperations = [];
     this.size = 0;
     this.sizeBytes = 0;
   }
@@ -538,12 +540,31 @@ async function executeCommands(
       }
     }
 
+    // The per-operation buffers were serialized in addToOperationsList using the
+    // bulk operation's BSON options. Reuse them only when the serialization
+    // options in effect for this execution still match, so that BSON options
+    // supplied to execute() (e.g. `ignoreUndefined`) are honored. Also skip reuse
+    // under auto-encryption, where libmongocrypt requires the plaintext command
+    // (document sequences are not permitted per the bulkWrite spec).
+    const createdBsonOptions = bulkOperation.s.bsonOptions;
+    const canReuseSerialized =
+      !bulkOperation.s.usingAutoEncryption &&
+      finalOptions.ignoreUndefined === createdBsonOptions.ignoreUndefined &&
+      finalOptions.serializeFunctions === createdBsonOptions.serializeFunctions &&
+      (finalOptions.checkKeys ?? false) === bulkOperation.s.checkKeys;
+    const serialized = canReuseSerialized ? batch.serializedOperations : undefined;
+
     const operation = isInsertBatch(batch)
-      ? new InsertOperation(bulkOperation.s.namespace, batch.operations, finalOptions)
+      ? new InsertOperation(bulkOperation.s.namespace, batch.operations, finalOptions, serialized)
       : isUpdateBatch(batch)
-        ? new UpdateOperation(bulkOperation.s.namespace, batch.operations, finalOptions)
+        ? new UpdateOperation(bulkOperation.s.namespace, batch.operations, finalOptions, serialized)
         : isDeleteBatch(batch)
-          ? new DeleteOperation(bulkOperation.s.namespace, batch.operations, finalOptions)
+          ? new DeleteOperation(
+              bulkOperation.s.namespace,
+              batch.operations,
+              finalOptions,
+              serialized
+            )
           : null;
 
     if (operation == null) throw new MongoRuntimeError(`Unknown batchType: ${batch.batchType}`);
@@ -560,6 +581,15 @@ async function executeCommands(
       thrownError = error;
     }
 
+    // Release this batch's serialized buffers now that it has been sent. The
+    // operation keeps its own reference to the array for the duration of its
+    // execution and any retries, so reassigning here (rather than mutating the
+    // shared array) only drops the bulk operation's copy. All batches are built
+    // up front, so this does not lower the peak (reached before the first send);
+    // it frees each batch's buffers as that batch completes so they are not kept
+    // alive through the remaining batches' round trips.
+    batch.serializedOperations = [];
+
     if (thrownError != null) {
       if (thrownError instanceof MongoWriteConcernError) {
         mergeBatchResults(batch, bulkOperation.s.bulkResult, thrownError, result);
@@ -824,6 +854,7 @@ export interface BulkOperationPrivate {
   // check keys
   checkKeys: boolean;
   bypassDocumentValidation?: boolean;
+  usingAutoEncryption: boolean;
 }
 
 /** @public */
@@ -953,7 +984,8 @@ export abstract class BulkOperationBase {
       // Fundamental error
       err: undefined,
       // check keys
-      checkKeys: typeof options.checkKeys === 'boolean' ? options.checkKeys : false
+      checkKeys: typeof options.checkKeys === 'boolean' ? options.checkKeys : false,
+      usingAutoEncryption
     };
 
     // bypass Validation

src/bulk/ordered.ts

@@ -17,13 +17,28 @@ export class OrderedBulkOperation extends BulkOperationBase {
     batchType: BatchType,
     document: Document | UpdateStatement | DeleteStatement
   ): this {
-    // Get the bsonSize
-    const bsonSize = BSON.calculateObjectSize(document, {
-      checkKeys: false,
-      // Since we don't know what the user selected for BSON options here,
-      // err on the safe side, and check the size with ignoreUndefined: false.
-      ignoreUndefined: false
-    } as any);
+    // Serialize the operation once here and reuse the bytes for both the size
+    // check/splitting and the wire message (via a DocumentSequence). Under
+    // auto-encryption the command is sent as a BSON array rather than a
+    // document sequence, so the buffer would never be reused; there we only
+    // measure the size and leave `buffer` undefined so nothing is retained on
+    // the batch.
+    let buffer: Uint8Array | undefined;
+    let bsonSize: number;
+    if (this.s.usingAutoEncryption) {
+      bsonSize = BSON.calculateObjectSize(document, {
+        checkKeys: false,
+        ignoreUndefined: false
+      } as any);
+    } else {
+      const bson = this.s.bsonOptions;
+      buffer = BSON.serialize(document, {
+        checkKeys: this.s.checkKeys,
+        ignoreUndefined: bson.ignoreUndefined,
+        serializeFunctions: bson.serializeFunctions
+      });
+      bsonSize = buffer.length;
+    }
 
     // Throw error if the doc is bigger than the max BSON size
     if (bsonSize >= this.s.maxBsonObjectSize)
@@ -75,6 +90,7 @@ export class OrderedBulkOperation extends BulkOperationBase {
 
     this.s.currentBatch.originalIndexes.push(this.s.currentIndex);
     this.s.currentBatch.operations.push(document);
+    if (buffer != null) this.s.currentBatch.serializedOperations.push(buffer);
     this.s.currentBatchSize += 1;
     this.s.currentBatchSizeBytes += maxKeySize + bsonSize;
     this.s.currentIndex += 1;

src/bulk/unordered.ts

@@ -31,14 +31,28 @@ export class UnorderedBulkOperation extends BulkOperationBase {
     batchType: BatchType,
     document: Document | UpdateStatement | DeleteStatement
   ): this {
-    // Get the bsonSize
-    const bsonSize = BSON.calculateObjectSize(document, {
-      checkKeys: false,
-
-      // Since we don't know what the user selected for BSON options here,
-      // err on the safe side, and check the size with ignoreUndefined: false.
-      ignoreUndefined: false
-    } as any);
+    // Serialize the operation once here and reuse the bytes for both the size
+    // check/splitting and the wire message (via a DocumentSequence). Under
+    // auto-encryption the command is sent as a BSON array rather than a
+    // document sequence, so the buffer would never be reused; there we only
+    // measure the size and leave `buffer` undefined so nothing is retained on
+    // the batch.
+    let buffer: Uint8Array | undefined;
+    let bsonSize: number;
+    if (this.s.usingAutoEncryption) {
+      bsonSize = BSON.calculateObjectSize(document, {
+        checkKeys: false,
+        ignoreUndefined: false
+      } as any);
+    } else {
+      const bson = this.s.bsonOptions;
+      buffer = BSON.serialize(document, {
+        checkKeys: this.s.checkKeys,
+        ignoreUndefined: bson.ignoreUndefined,
+        serializeFunctions: bson.serializeFunctions
+      });
+      bsonSize = buffer.length;
+    }
 
     // Throw error if the doc is bigger than the max BSON size
     if (bsonSize >= this.s.maxBsonObjectSize) {
@@ -90,6 +104,7 @@ export class UnorderedBulkOperation extends BulkOperationBase {
     }
 
     this.s.currentBatch.operations.push(document);
+    if (buffer != null) this.s.currentBatch.serializedOperations.push(buffer);
     this.s.currentBatch.originalIndexes.push(this.s.currentIndex);
     this.s.currentIndex = this.s.currentIndex + 1;
 

src/cmap/command_monitoring_events.ts

@@ -249,13 +249,12 @@ const LEGACY_FIND_OPTIONS_MAP = {
 function extractCommand(command: WriteProtocolMessageType): Document {
   if (command instanceof OpMsgRequest) {
     const cmd = { ...command.command };
-    // For OP_MSG with payload type 1 we need to pull the documents
-    // array out of the document sequence for monitoring.
-    if (cmd.ops instanceof DocumentSequence) {
-      cmd.ops = cmd.ops.documents;
-    }
-    if (cmd.nsInfo instanceof DocumentSequence) {
-      cmd.nsInfo = cmd.nsInfo.documents;
+    // For OP_MSG payload type 1, replace any document-sequence field with its
+    // documents array for monitoring (ops, nsInfo, documents, updates, deletes).
+    for (const key of Object.keys(cmd)) {
+      if (cmd[key] instanceof DocumentSequence) {
+        cmd[key] = cmd[key].documents;
+      }
     }
     return cmd;
   }

src/cmap/commands.ts

@@ -500,6 +500,23 @@ export class DocumentSequence {
   }
 }
 
+/**
+ * Build a DocumentSequence from documents that have already been serialized,
+ * reusing the provided buffers instead of re-serializing.
+ * @internal
+ */
+export function makeDocumentSequence(
+  field: string,
+  documents: Document[],
+  serialized: Uint8Array[]
+): DocumentSequence {
+  const sequence = new DocumentSequence(field);
+  for (let i = 0; i < documents.length; i++) {
+    sequence.push(documents[i], serialized[i]);
+  }
+  return sequence;
+}
+
 /** @internal */
 export class OpMsgRequest {
   requestId: number;

src/operations/delete.ts

@@ -1,4 +1,5 @@
 import type { Document } from '../bson';
+import { makeDocumentSequence } from '../cmap/commands';
 import { type Connection } from '../cmap/connection';
 import { MongoDBResponse } from '../cmap/wire_protocol/responses';
 import { MongoCompatibilityError, MongoServerError } from '../error';
@@ -45,12 +46,20 @@ export class DeleteOperation extends CommandOperation<Document> {
   override SERVER_COMMAND_RESPONSE_TYPE = MongoDBResponse;
   override options: DeleteOptions;
   statements: DeleteStatement[];
-
-  constructor(ns: MongoDBNamespace, statements: DeleteStatement[], options: DeleteOptions) {
+  /** @internal */
+  serializedOperations?: Uint8Array[];
+
+  constructor(
+    ns: MongoDBNamespace,
+    statements: DeleteStatement[],
+    options: DeleteOptions,
+    serializedOperations?: Uint8Array[]
+  ) {
     super(undefined, options);
     this.options = options;
     this.ns = ns;
     this.statements = statements;
+    this.serializedOperations = serializedOperations;
   }
 
   override get commandName() {
@@ -71,7 +80,9 @@ export class DeleteOperation extends CommandOperation<Document> {
     const ordered = typeof options.ordered === 'boolean' ? options.ordered : true;
     const command: Document = {
       delete: this.ns.collection,
-      deletes: this.statements,
+      deletes: this.serializedOperations
+        ? makeDocumentSequence('deletes', this.statements, this.serializedOperations)
+        : this.statements,
       ordered
     };
 

src/operations/insert.ts

@@ -1,6 +1,7 @@
 import { type Connection } from '..';
 import type { Document } from '../bson';
 import type { BulkWriteOptions } from '../bulk/common';
+import { makeDocumentSequence } from '../cmap/commands';
 import { MongoDBResponse } from '../cmap/wire_protocol/responses';
 import type { Collection } from '../collection';
 import { MongoServerError } from '../error';
@@ -15,12 +16,20 @@ export class InsertOperation extends CommandOperation<Document> {
   override options: BulkWriteOptions;
 
   documents: Document[];
+  /** @internal Per-operation pre-serialized BSON, reused to build a DocumentSequence. */
+  serializedOperations?: Uint8Array[];
 
-  constructor(ns: MongoDBNamespace, documents: Document[], options: BulkWriteOptions) {
+  constructor(
+    ns: MongoDBNamespace,
+    documents: Document[],
+    options: BulkWriteOptions,
+    serializedOperations?: Uint8Array[]
+  ) {
     super(undefined, options);
     this.options = { ...options, checkKeys: options.checkKeys ?? false };
     this.ns = ns;
     this.documents = documents;
+    this.serializedOperations = serializedOperations;
   }
 
   override get commandName() {
@@ -32,7 +41,9 @@ export class InsertOperation extends CommandOperation<Document> {
     const ordered = typeof options.ordered === 'boolean' ? options.ordered : true;
     const command: Document = {
       insert: this.ns.collection,
-      documents: this.documents,
+      documents: this.serializedOperations
+        ? makeDocumentSequence('documents', this.documents, this.serializedOperations)
+        : this.documents,
       ordered
     };
 

src/operations/update.ts

@@ -1,4 +1,5 @@
 import type { Document } from '../bson';
+import { makeDocumentSequence } from '../cmap/commands';
 import { type Connection } from '../cmap/connection';
 import { MongoDBResponse } from '../cmap/wire_protocol/responses';
 import { MongoInvalidArgumentError, MongoServerError } from '../error';
@@ -74,17 +75,21 @@ export class UpdateOperation extends CommandOperation<Document> {
   override SERVER_COMMAND_RESPONSE_TYPE = MongoDBResponse;
   override options: UpdateOptions & { ordered?: boolean };
   statements: UpdateStatement[];
+  /** @internal */
+  serializedOperations?: Uint8Array[];
 
   constructor(
     ns: MongoDBNamespace,
     statements: UpdateStatement[],
-    options: UpdateOptions & { ordered?: boolean }
+    options: UpdateOptions & { ordered?: boolean },
+    serializedOperations?: Uint8Array[]
   ) {
     super(undefined, options);
     this.options = options;
     this.ns = ns;
 
     this.statements = statements;
+    this.serializedOperations = serializedOperations;
   }
 
   override get commandName() {
@@ -103,7 +108,9 @@ export class UpdateOperation extends CommandOperation<Document> {
     const options = this.options;
     const command: Document = {
       update: this.ns.collection,
-      updates: this.statements,
+      updates: this.serializedOperations
+        ? makeDocumentSequence('updates', this.statements, this.serializedOperations)
+        : this.statements,
       ordered: options.ordered ?? true
     };