feat: add Mongo bulkWrite API MONGOSH-1100

Author: gagik

packages/service-provider-core/src/all-transport-types.ts

@@ -2,9 +2,11 @@ export type {
   AggregateOptions,
   AggregationCursor,
   AnyBulkWriteOperation,
+  AnyClientBulkWriteModel,
   Batch,
   BulkWriteOptions,
   BulkWriteResult,
+  ClientBulkWriteResult,
   ChangeStream,
   ChangeStreamOptions,
   ClientSession,

packages/service-provider-core/src/writable.ts

@@ -1,4 +1,4 @@
-import type { RunCursorCommandOptions } from 'mongodb';
+import type { ClientBulkWriteOptions, RunCursorCommandOptions } from 'mongodb';
 import type {
   Document,
   InsertOneOptions,
@@ -10,6 +10,8 @@ import type {
   FindOneAndUpdateOptions,
   BulkWriteOptions,
   AnyBulkWriteOperation,
+  AnyClientBulkWriteModel,
+  ClientBulkWriteResult,
   DeleteOptions,
   DeleteResult,
   InsertManyResult,
@@ -108,6 +110,18 @@ export default interface Writable {
     dbOptions?: DbOptions
   ): Promise<BulkWriteResult>;
 
+  /**
+   * Executes a client bulk write operation, available on server 8.0+.
+   * @param models - The client bulk write models.
+   * @param options - The bulk write options.
+   *
+   * @returns {Promise} The promise of the result.
+   */
+  clientBulkWrite(
+    models: AnyClientBulkWriteModel<Document>[],
+    options: ClientBulkWriteOptions
+  ): Promise<ClientBulkWriteResult>;
+
   /**
    * Delete multiple documents from the collection.
    *

packages/service-provider-node-driver/src/node-driver-service-provider.ts

@@ -23,8 +23,10 @@ import type {
   AggregateOptions,
   AggregationCursor,
   AnyBulkWriteOperation,
+  AnyClientBulkWriteModel,
   BulkWriteOptions,
   BulkWriteResult,
+  ClientBulkWriteResult,
   ClientSessionOptions,
   Collection,
   CountDocumentsOptions,
@@ -630,6 +632,17 @@ export class NodeDriverServiceProvider
       .bulkWrite(requests, options);
   }
 
+  /**
+   * Executes a client bulk write operation, available on server 8.0+.
+   */
+  clientBulkWrite(
+    models: AnyClientBulkWriteModel<Document>[],
+    options: BulkWriteOptions = {}
+  ): Promise<ClientBulkWriteResult> {
+    options = { ...this.baseCmdOptions, ...options };
+    return this.mongoClient.bulkWrite(models, options);
+  }
+
   /**
    * Close the connection.
    *
@@ -997,7 +1010,7 @@ export class NodeDriverServiceProvider
     options = { ...this.baseCmdOptions, ...options };
     return this.db(database, dbOptions)
       .collection(collection)
-      .replaceOne(filter, replacement, options) as Promise<UpdateResult>;
+      .replaceOne(filter, replacement, options);
     // `as UpdateResult` because we know we didn't request .explain() here.
   }
 

packages/shell-api/src/mongo.spec.ts

@@ -1015,6 +1015,74 @@ describe('Mongo', function () {
           ).toArray();
         });
       });
+
+      context('post-8.0', function () {
+        skipIfServerVersion(testServer, '< 8.0');
+        let mongo: Mongo;
+
+        describe('bulkWrite', function () {
+          beforeEach(async function () {
+            mongo = await instanceState.shellApi.Mongo(uri, undefined, {
+              api: { version: '1' },
+            });
+          });
+
+          it('should allow inserts across collections and databases', async function () {
+            expect(
+              await mongo.bulkWrite([
+                {
+                  name: 'insertOne',
+                  namespace: 'db.authors',
+                  document: { name: 'King' },
+                },
+                {
+                  name: 'deleteOne',
+                  namespace: 'db.authors',
+                  filter: { name: 'King' },
+                },
+                {
+                  name: 'insertOne',
+                  namespace: 'db.moreAuthors',
+                  document: { name: 'Queen' },
+                },
+                {
+                  name: 'insertOne',
+                  namespace: 'otherDb.authors',
+                  document: { name: 'Prince' },
+                },
+              ])
+            ).deep.equals({
+              acknowledged: true,
+              insertedCount: 3,
+              upsertedCount: 0,
+              matchedCount: 0,
+              modifiedCount: 0,
+              deletedCount: 1,
+              insertResults: undefined,
+              updateResults: undefined,
+              deleteResults: undefined,
+            });
+
+            expect(
+              await mongo.getDB('db').getCollection('authors').count()
+            ).equals(0);
+
+            expect(
+              await mongo
+                .getDB('db')
+                .getCollection('moreAuthors')
+                .count({ name: 'Queen' })
+            ).equals(1);
+
+            expect(
+              await mongo
+                .getDB('otherDb')
+                .getCollection('authors')
+                .count({ name: 'Prince' })
+            ).equals(1);
+          });
+        });
+      });
     });
   });
 });

packages/shell-api/src/mongo.ts

@@ -36,6 +36,8 @@ import type {
   ServerApi,
   ServerApiVersion,
   WriteConcern,
+  AnyClientBulkWriteModel,
+  ClientBulkWriteResult,
 } from '@mongosh/service-provider-core';
 import type { ConnectionInfo } from '@mongosh/arg-parser';
 import {
@@ -62,6 +64,7 @@ import { ShellApiErrors } from './error-codes';
 import type { LogEntry } from './log-entry';
 import { parseAnyLogEntry } from './log-entry';
 import type { ShellBson } from './shell-bson';
+import type { ClientBulkWriteOptions } from 'mongodb';
 
 /* Utility, inverse of Readonly<T> */
 type Mutable<T> = {
@@ -365,6 +368,18 @@ export default class Mongo extends ShellApiClass {
     return await this._listDatabases(options);
   }
 
+  @returnsPromise
+  @serverVersions(['8.0.0', ServerVersions.latest])
+  @apiVersions([1])
+  bulkWrite(
+    models: AnyClientBulkWriteModel<Document>[],
+    options: ClientBulkWriteOptions = {}
+  ): Promise<ClientBulkWriteResult> {
+    this._emitMongoApiCall('bulkWrite', { options });
+
+    return this._serviceProvider.clientBulkWrite(models, options);
+  }
+
   @returnsPromise
   @apiVersions([1])
   async getDBNames(options: ListDatabasesOptions = {}): Promise<string[]> {