All aworking

Author: baileympearson

src/bulk/common.ts

@@ -14,13 +14,11 @@ import type { CollationOptions, CommandOperationOptions } from '../operations/co
 import { DeleteOperation, type DeleteStatement, makeDeleteStatement } from '../operations/delete';
 import { executeOperation } from '../operations/execute_operation';
 import { InsertOperation } from '../operations/insert';
-import { AbstractOperation, type Hint } from '../operations/operation';
+import { type Hint } from '../operations/operation';
 import { makeUpdateStatement, UpdateOperation, type UpdateStatement } from '../operations/update';
-import type { Server } from '../sdam/server';
 import type { Topology } from '../sdam/topology';
-import type { ClientSession } from '../sessions';
 import { type Sort } from '../sort';
-import { type TimeoutContext } from '../timeout';
+import { TimeoutContext } from '../timeout';
 import {
   applyRetryableWrites,
   getTopology,
@@ -854,40 +852,6 @@ export interface BulkWriteOptions extends CommandOperationOptions {
   timeoutContext?: TimeoutContext;
 }
 
-/**
- * TODO(NODE-4063)
- * BulkWrites merge complexity is implemented in executeCommands
- * This provides a vehicle to treat bulkOperations like any other operation (hence "shim")
- * We would like this logic to simply live inside the BulkWriteOperation class
- * @internal
- */
-export class BulkWriteShimOperation extends AbstractOperation {
-  bulkOperation: BulkOperationBase;
-  constructor(bulkOperation: BulkOperationBase, options: BulkWriteOptions) {
-    super(options);
-    this.bulkOperation = bulkOperation;
-  }
-
-  get commandName(): string {
-    return 'bulkWrite' as const;
-  }
-
-  async execute(
-    _server: Server,
-    session: ClientSession | undefined,
-    timeoutContext: TimeoutContext
-  ): Promise<any> {
-    if (this.options.session == null) {
-      // An implicit session could have been created by 'executeOperation'
-      // So if we stick it on finalOptions here, each bulk operation
-      // will use this same session, it'll be passed in the same way
-      // an explicit session would be
-      this.options.session = session;
-    }
-    return await executeCommands(this.bulkOperation, { ...this.options, timeoutContext });
-  }
-}
-
 /** @public */
 export abstract class BulkOperationBase {
   isOrdered: boolean;
@@ -1208,10 +1172,26 @@ export abstract class BulkOperationBase {
     }
 
     this.s.executed = true;
-    const finalOptions = { ...this.s.options, ...options };
-    const operation = new BulkWriteShimOperation(this, finalOptions);
+    const finalOptions = resolveOptions(this.collection, { ...this.s.options, ...options });
+
+    // if there is no timeoutContext provided, create a timeoutContext and use it for
+    // all batches in the bulk operation
+    finalOptions.timeoutContext ??= TimeoutContext.create({
+      session: finalOptions.session,
+      timeoutMS: finalOptions.timeoutMS,
+      serverSelectionTimeoutMS: this.collection.client.s.options.serverSelectionTimeoutMS,
+      waitQueueTimeoutMS: this.collection.client.s.options.waitQueueTimeoutMS
+    });
+
+    if (finalOptions.session == null) {
+      // if there is not an explicit session provided to `execute()`, create
+      // an implicit session and use that for all batches in the bulk operation
+      return await this.collection.client.withSession({ explicit: false }, async session => {
+        return await executeCommands(this, { ...finalOptions, session });
+      });
+    }
 
-    return await executeOperation(this.s.collection.client, operation, finalOptions.timeoutContext);
+    return await executeCommands(this, { ...finalOptions });
   }
 
   /**

src/collection.ts

@@ -42,7 +42,7 @@ import {
   EstimatedDocumentCountOperation,
   type EstimatedDocumentCountOptions
 } from './operations/estimated_document_count';
-import { executeOperation } from './operations/execute_operation';
+import { autoConnect, executeOperation } from './operations/execute_operation';
 import type { FindOptions } from './operations/find';
 import {
   FindOneAndDeleteOperation,
@@ -311,7 +311,7 @@ export class Collection<TSchema extends Document = Document> {
     if (!Array.isArray(docs)) {
       throw new MongoInvalidArgumentError('Argument "docs" must be an array of documents');
     }
-    options = options ?? {};
+    options = resolveOptions(this, options ?? {});
 
     const acknowledged = WriteConcern.fromOptions(options)?.w !== 0;
 
@@ -362,7 +362,14 @@ export class Collection<TSchema extends Document = Document> {
       throw new MongoInvalidArgumentError('Argument "operations" must be an array of documents');
     }
 
-    options = options ?? {};
+    options = resolveOptions(this, options ?? {});
+
+    // TODO(NODE-7071): remove once the client doesn't need to be connected to construct
+    // bulk operations
+    const isConnected = this.client.topology != null;
+    if (!isConnected) {
+      await autoConnect(this.client);
+    }
 
     // Create the bulk operation
     const bulk: BulkOperationBase =
@@ -371,8 +378,8 @@ export class Collection<TSchema extends Document = Document> {
         : this.initializeOrderedBulkOp(options);
 
     // for each op go through and add to the bulk
-    for (let i = 0; i < operations.length; i++) {
-      bulk.raw(operations[i]);
+    for (const operation of operations) {
+      bulk.raw(operation);
     }
 
     // Execute the bulk

src/operations/execute_operation.ts

@@ -129,7 +129,7 @@ export async function executeOperation<
  * Connects a client if it has not yet been connected
  * @internal
  */
-async function autoConnect(client: MongoClient): Promise<Topology> {
+export async function autoConnect(client: MongoClient): Promise<Topology> {
   if (client.topology == null) {
     if (client.s.hasBeenClosed) {
       throw new MongoNotConnectedError('Client must be connected before running operations');

test/integration/crud/bulk.test.ts

@@ -925,7 +925,7 @@ describe('Bulk', function () {
         try {
           batch.insert({ string: hugeString });
           test.ok(false);
-        } catch (err) {} // eslint-disable-line
+        } catch (err) { } // eslint-disable-line
 
         // Finish up test
         client.close(done);
@@ -1216,34 +1216,27 @@ describe('Bulk', function () {
     }
   });
 
-  it('should correctly execute unordered batch using w:0', {
-    metadata: { requires: { topology: ['single', 'replicaset', 'ssl', 'heap', 'wiredtiger'] } },
+  it('should correctly execute unordered batch using w:0', async function () {
+    await client.connect();
+    const db = client.db();
+    const col = db.collection('batch_write_ordered_ops_9');
+    const bulk = col.initializeUnorderedBulkOp();
+    for (let i = 0; i < 100; i++) {
+      bulk.insert({ a: 1 });
+    }
 
-    test: function (done) {
-      client.connect((err, client) => {
-        const db = client.db();
-        const col = db.collection('batch_write_ordered_ops_9');
-        const bulk = col.initializeUnorderedBulkOp();
-        for (let i = 0; i < 100; i++) {
-          bulk.insert({ a: 1 });
-        }
+    bulk.find({ b: 1 }).upsert().update({ b: 1 });
+    bulk.find({ c: 1 }).delete();
 
-        bulk.find({ b: 1 }).upsert().update({ b: 1 });
-        bulk.find({ c: 1 }).delete();
+    const result = await bulk.execute({ writeConcern: { w: 0 } });
+    test.equal(0, result.upsertedCount);
+    test.equal(0, result.insertedCount);
+    test.equal(0, result.matchedCount);
+    test.ok(0 === result.modifiedCount || result.modifiedCount == null);
+    test.equal(0, result.deletedCount);
+    test.equal(false, result.hasWriteErrors());
 
-        bulk.execute({ writeConcern: { w: 0 } }, function (err, result) {
-          expect(err).to.not.exist;
-          test.equal(0, result.upsertedCount);
-          test.equal(0, result.insertedCount);
-          test.equal(0, result.matchedCount);
-          test.ok(0 === result.modifiedCount || result.modifiedCount == null);
-          test.equal(0, result.deletedCount);
-          test.equal(false, result.hasWriteErrors());
-
-          client.close(done);
-        });
-      });
-    }
+    await client.close();
   });
 
   it('should provide an accessor for operations on ordered bulk ops', function (done) {

test/mongodb.ts

@@ -163,7 +163,6 @@ export * from '../src/mongo_client';
 export * from '../src/mongo_logger';
 export * from '../src/mongo_types';
 export * from '../src/operations/aggregate';
-export * from '../src/operations/bulk_write';
 export * from '../src/operations/client_bulk_write/command_builder';
 export * from '../src/operations/client_bulk_write/common';
 export * from '../src/operations/client_bulk_write/results_merger';