diff --git a/src/bulk/common.ts b/src/bulk/common.ts index 26c0d750d9c..77727cd9438 100644 --- a/src/bulk/common.ts +++ b/src/bulk/common.ts @@ -157,6 +157,7 @@ export class Batch { originalIndexes: number[]; batchType: BatchType; operations: T[]; + serializedOperations: Uint8Array[]; size: number; sizeBytes: number; @@ -166,6 +167,7 @@ export class Batch { 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 diff --git a/src/bulk/ordered.ts b/src/bulk/ordered.ts index 2a9e9ece2b1..4aa0acb5a4d 100644 --- a/src/bulk/ordered.ts +++ b/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; diff --git a/src/bulk/unordered.ts b/src/bulk/unordered.ts index 97a134613b0..c718377722a 100644 --- a/src/bulk/unordered.ts +++ b/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; diff --git a/src/cmap/command_monitoring_events.ts b/src/cmap/command_monitoring_events.ts index b5e7c776d57..437ecd7b920 100644 --- a/src/cmap/command_monitoring_events.ts +++ b/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; } diff --git a/src/cmap/commands.ts b/src/cmap/commands.ts index bdc332cbdf1..1a5c9b7f4e8 100644 --- a/src/cmap/commands.ts +++ b/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; diff --git a/src/operations/delete.ts b/src/operations/delete.ts index 94b6435e1bb..d27afc664be 100644 --- a/src/operations/delete.ts +++ b/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 { 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 { 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 }; diff --git a/src/operations/insert.ts b/src/operations/insert.ts index 45d470bf05c..c7c4b77fb48 100644 --- a/src/operations/insert.ts +++ b/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 { 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 { 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 }; diff --git a/src/operations/update.ts b/src/operations/update.ts index 36d7a5e3c78..96623f3263d 100644 --- a/src/operations/update.ts +++ b/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 { 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 { 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 }; diff --git a/test/integration/crud/bulk.test.ts b/test/integration/crud/bulk.test.ts index 0f8ae3ae60f..5390babe165 100644 --- a/test/integration/crud/bulk.test.ts +++ b/test/integration/crud/bulk.test.ts @@ -4,6 +4,7 @@ import { expect } from 'chai'; import { type Collection, + type CommandStartedEvent, Double, Long, MongoBatchReExecutionError, @@ -1860,4 +1861,98 @@ describe('Bulk', function () { } }); }); + + describe('#insertMany', function () { + context('when a document field is set to undefined', function () { + it('inserts the field as null by default (ignoreUndefined defaults to false)', async function () { + const collection = client + .db(DB_NAME) + .collection<{ _id: number; a?: null }>('undefined_fields'); + await collection.insertMany([{ _id: 1, a: undefined }]); + + const doc = await collection.findOne({ _id: 1 }); + expect(doc).to.have.property('a', null); + }); + }); + + context('when monitoring commands', function () { + it('reports the documents field of the insert command as an array', async function () { + const collection = client.db(DB_NAME).collection<{ _id: number }>('command_events'); + const commands: CommandStartedEvent[] = []; + client.on('commandStarted', event => { + if (event.commandName === 'insert') commands.push(event); + }); + + await collection.insertMany([{ _id: 1 }, { _id: 2 }]); + + expect(commands).to.have.lengthOf(1); + expect(commands[0].command.documents).to.be.an('array').with.lengthOf(2); + }); + }); + + context('when the operation has been executed', function () { + it('releases the serialized operation buffers held on each batch', async function () { + const collection = client.db(DB_NAME).collection<{ _id: number }>('buffer_release'); + const bulk = collection.initializeOrderedBulkOp(); + bulk.insert({ _id: 1 }); + bulk.insert({ _id: 2 }); + + // The getter returns a copy of the batches array, but the Batch objects + // themselves are the same instances the driver executes and survive the + // internal batches array being cleared after execution. + const batches = bulk.batches; + expect(batches).to.have.lengthOf(1); + expect(batches[0].serializedOperations).to.have.lengthOf(2); + + await bulk.execute(); + + expect(batches[0].serializedOperations).to.have.lengthOf(0); + }); + }); + }); + + describe('BSON options passed to #execute', function () { + // Documents are serialized once when added to the bulk operation, then the + // resulting bytes are reused as the wire command. BSON options supplied to + // execute() must still take effect, which requires falling back to + // re-serialization when they differ from the options used when the + // documents were added. + for (const ordered of [true, false]) { + context(`when the bulk operation is ${ordered ? 'ordered' : 'unordered'}`, function () { + context('when execute() overrides ignoreUndefined to true', function () { + it('omits undefined fields instead of writing them as null', async function () { + const collection = client + .db(DB_NAME) + .collection<{ _id: number; a?: null }>('execute_bson_options'); + const bulk = ordered + ? collection.initializeOrderedBulkOp() + : collection.initializeUnorderedBulkOp(); + bulk.insert({ _id: 1, a: undefined }); + + await bulk.execute({ ignoreUndefined: true }); + + const doc = await collection.findOne({ _id: 1 }); + expect(doc).to.not.have.property('a'); + }); + }); + + context('when execute() does not override BSON options', function () { + it('serializes the undefined field as null (ignoreUndefined defaults to false)', async function () { + const collection = client + .db(DB_NAME) + .collection<{ _id: number; a?: null }>('execute_bson_options_default'); + const bulk = ordered + ? collection.initializeOrderedBulkOp() + : collection.initializeUnorderedBulkOp(); + bulk.insert({ _id: 1, a: undefined }); + + await bulk.execute(); + + const doc = await collection.findOne({ _id: 1 }); + expect(doc).to.have.property('a', null); + }); + }); + }); + } + }); }); diff --git a/test/unit/assorted/collations.test.js b/test/unit/assorted/collations.test.js index 8a245a0288d..c02ebf208bc 100644 --- a/test/unit/assorted/collations.test.js +++ b/test/unit/assorted/collations.test.js @@ -347,9 +347,15 @@ describe('Collation', function () { .then(() => { expect(commandResult).to.exist; expect(commandResult).to.have.property('updates'); - expect(commandResult.updates).to.have.length.at.least(1); - expect(commandResult.updates[0]).to.have.property('collation'); - expect(commandResult.updates[0].collation).to.eql({ caseLevel: true }); + // `updates` is sent as an OP_MSG type-1 document sequence rather than an + // inline BSON array, so the mock server surfaces it as a DocumentSequence-like + // object with a `documents` array instead of a plain array. + const updates = Array.isArray(commandResult.updates) + ? commandResult.updates + : commandResult.updates.documents; + expect(updates).to.have.length.at.least(1); + expect(updates[0]).to.have.property('collation'); + expect(updates[0].collation).to.eql({ caseLevel: true }); return client.close(); }); diff --git a/test/unit/cmap/command_monitoring_events.test.js b/test/unit/cmap/command_monitoring_events.test.js index ae5dcfa57ed..ca310ae1470 100644 --- a/test/unit/cmap/command_monitoring_events.test.js +++ b/test/unit/cmap/command_monitoring_events.test.js @@ -1,6 +1,11 @@ 'use strict'; -const { OpQueryRequest, OpMsgRequest, CommandStartedEvent } = require('../../mongodb'); +const { + OpQueryRequest, + OpMsgRequest, + CommandStartedEvent, + DocumentSequence +} = require('../../mongodb'); const { expect } = require('chai'); describe('Command Monitoring Events - unit/cmap', function () { @@ -90,4 +95,30 @@ describe('Command Monitoring Events - unit/cmap', function () { expect(startEvent).to.have.property('command').that.deep.equals(query.query.$query); }); }); + + for (const { name, field, documents } of [ + { name: 'insert', field: 'documents', documents: [{ _id: 1 }, { _id: 2 }] }, + { name: 'update', field: 'updates', documents: [{ q: { _id: 1 }, u: { $set: { a: 1 } } }] }, + { name: 'delete', field: 'deletes', documents: [{ q: { _id: 1 }, limit: 1 }] } + ]) { + it(`reconstructs the ${field} document sequence into an array (${name} command)`, function () { + const command = { + [name]: 'coll', + [field]: new DocumentSequence(field, documents), + $db: 'test' + }; + const msg = new OpMsgRequest('test', command, {}); + const event = new CommandStartedEvent({ id: 1, address: '127.0.0.1:27017' }, msg); + expect(event.command[field]).to.be.an('array').that.deep.equals(documents); + }); + } + + it('leaves non-document-sequence fields untouched', function () { + const pipeline = [{ $match: { a: 1 } }]; + const command = { aggregate: 'coll', pipeline, cursor: {}, $db: 'test' }; + const msg = new OpMsgRequest('test', command, {}); + const event = new CommandStartedEvent({ id: 1, address: '127.0.0.1:27017' }, msg); + expect(event.command.pipeline).to.be.an('array').that.deep.equals(pipeline); + expect(event.command.cursor).to.deep.equal({}); + }); }); diff --git a/test/unit/operations/insert.test.ts b/test/unit/operations/insert.test.ts new file mode 100644 index 00000000000..8e18f0e0b0b --- /dev/null +++ b/test/unit/operations/insert.test.ts @@ -0,0 +1,24 @@ +import * as BSON from 'bson'; +import { expect } from 'chai'; + +import { DocumentSequence, InsertOperation, MongoDBNamespace } from '../../mongodb'; + +describe('InsertOperation', function () { + const ns = MongoDBNamespace.fromString('test.coll'); + const docs = [{ _id: 1 }, { _id: 2 }]; + + it('uses a DocumentSequence when serialized buffers are provided', function () { + const buffers = docs.map(d => BSON.serialize(d)); + const op = new InsertOperation(ns, docs, {}, buffers); + const command = op.buildCommandDocument({} as any); + expect(command.documents).to.be.instanceOf(DocumentSequence); + expect(command.documents.documents).to.deep.equal(docs); + }); + + it('uses a plain array when no serialized buffers are provided', function () { + const op = new InsertOperation(ns, docs, {}); + const command = op.buildCommandDocument({} as any); + expect(Array.isArray(command.documents)).to.equal(true); + expect(command.documents).to.deep.equal(docs); + }); +}); diff --git a/test/unit/operations/write_ops.test.ts b/test/unit/operations/write_ops.test.ts new file mode 100644 index 00000000000..b80a2d65e0a --- /dev/null +++ b/test/unit/operations/write_ops.test.ts @@ -0,0 +1,38 @@ +import * as BSON from 'bson'; +import { expect } from 'chai'; + +import { + DeleteOperation, + DocumentSequence, + MongoDBNamespace, + UpdateOperation +} from '../../mongodb'; + +describe('Update/Delete DocumentSequence', function () { + const ns = MongoDBNamespace.fromString('test.coll'); + + it('UpdateOperation uses a DocumentSequence for updates when buffers provided', function () { + const statements = [{ q: { a: 1 }, u: { $set: { b: 2 } } }]; + const buffers = statements.map(s => BSON.serialize(s)); + const op = new UpdateOperation(ns, statements as any, {}, buffers); + const command = op.buildCommandDocument({} as any); + expect(command.updates).to.be.instanceOf(DocumentSequence); + expect(command.updates.documents).to.deep.equal(statements); + }); + + it('DeleteOperation uses a DocumentSequence for deletes when buffers provided', function () { + const statements = [{ q: { a: 1 }, limit: 1 }]; + const buffers = statements.map(s => BSON.serialize(s)); + const op = new DeleteOperation(ns, statements as any, {}, buffers); + const command = op.buildCommandDocument({} as any); + expect(command.deletes).to.be.instanceOf(DocumentSequence); + expect(command.deletes.documents).to.deep.equal(statements); + }); + + it('falls back to arrays when no buffers provided', function () { + const u = new UpdateOperation(ns, [{ q: {}, u: {} }] as any, {}); + const d = new DeleteOperation(ns, [{ q: {}, limit: 1 }] as any, {}); + expect(Array.isArray(u.buildCommandDocument({} as any).updates)).to.equal(true); + expect(Array.isArray(d.buildCommandDocument({} as any).deletes)).to.equal(true); + }); +});