feat(bulk)!: add collation to FindOperators (#2679)

The primary purpose of this PR was to add a fluent builder method
for `collation` for bulk writes with a filter. As part of this work,
legacy support for raw operations was removed from
`collection.bulkWrite`.
A TypeError will now be thrown if attempting to bulkWrite with raw
operations.

NODE-2757

src/collection.ts

@@ -333,6 +333,8 @@ export class Collection {
    * ```js
    *  { insertOne: { document: { a: 1 } } }
    *
+   *  { insertMany: [{ g: 1 }, { g: 2 }]}
+   *
    *  { updateOne: { filter: {a:2}, update: {$set: {a:2}}, upsert:true } }
    *
    *  { updateMany: { filter: {a:2}, update: {$set: {a:2}}, upsert:true } }
@@ -345,6 +347,7 @@ export class Collection {
    *
    *  { replaceOne: { filter: {c:3}, replacement: {c:4}, upsert:true}}
    *```
+   * Please note that raw operations are no longer accepted as of driver version 4.0.
    *
    * If documents passed in do not contain the **_id** field,
    * one will be added to each of the documents missing it by the driver, mutating the document. This behavior

src/operations/delete.ts

@@ -1,6 +1,6 @@
 import { defineAspects, Aspect, Hint } from './operation';
 import { CommandOperation, CommandOperationOptions, CollationOptions } from './command';
-import { Callback, maxWireVersion, MongoDBNamespace } from '../utils';
+import { Callback, maxWireVersion, MongoDBNamespace, collationNotSupported } from '../utils';
 import type { Document } from '../bson';
 import type { Server } from '../sdam/server';
 import type { Collection } from '../collection';
@@ -88,6 +88,12 @@ export class DeleteOperation extends CommandOperation<Document> {
       }
     }
 
+    const statementWithCollation = this.statements.find(statement => !!statement.collation);
+    if (statementWithCollation && collationNotSupported(server, statementWithCollation)) {
+      callback(new MongoError(`server ${server.name} does not support collation`));
+      return;
+    }
+
     super.executeCommand(server, session, command, callback);
   }
 }
@@ -132,7 +138,7 @@ export class DeleteManyOperation extends DeleteOperation {
   }
 }
 
-function makeDeleteStatement(
+export function makeDeleteStatement(
   filter: Document,
   options: DeleteOptions & { limit?: number }
 ): DeleteStatement {

src/operations/update.ts

@@ -97,7 +97,11 @@ export class UpdateOperation extends CommandOperation<Document> {
       command.bypassDocumentValidation = options.bypassDocumentValidation;
     }
 
-    if (collationNotSupported(server, options)) {
+    const statementWithCollation = this.statements.find(statement => !!statement.collation);
+    if (
+      collationNotSupported(server, options) ||
+      (statementWithCollation && collationNotSupported(server, statementWithCollation))
+    ) {
       callback(new MongoError(`server ${server.name} does not support collation`));
       return;
     }
@@ -115,6 +119,11 @@ export class UpdateOperation extends CommandOperation<Document> {
       return;
     }
 
+    if (this.statements.some(statement => !!statement.arrayFilters) && maxWireVersion(server) < 6) {
+      callback(new MongoError('arrayFilters are only supported on MongoDB 3.6+'));
+      return;
+    }
+
     super.executeCommand(server, session, command, callback);
   }
 }
@@ -247,7 +256,7 @@ export class ReplaceOneOperation extends UpdateOperation {
   }
 }
 
-function makeUpdateStatement(
+export function makeUpdateStatement(
   filter: Document,
   update: Document,
   options: UpdateOptions & { multi?: boolean }

test/functional/apm.test.js

@@ -483,9 +483,9 @@ describe('APM', function () {
           .collection('apm_test_3')
           .bulkWrite(
             [
-              { insertOne: { a: 1 } },
-              { updateOne: { q: { a: 2 }, u: { $set: { a: 2 } }, upsert: true } },
-              { deleteOne: { q: { c: 1 } } }
+              { insertOne: { document: { a: 1 } } },
+              { updateOne: { filter: { a: 2 }, update: { $set: { a: 2 } }, upsert: true } },
+              { deleteOne: { filter: { c: 1 } } }
             ],
             { ordered: true }
           )

test/functional/bulk.test.js

@@ -1,5 +1,11 @@
 'use strict';
-const { withClient, withClientV2, setupDatabase, ignoreNsNotFound } = require('./shared');
+const {
+  withClient,
+  withClientV2,
+  withMonitoredClient,
+  setupDatabase,
+  ignoreNsNotFound
+} = require('./shared');
 const test = require('./shared').assert;
 const { MongoError } = require('../../src/error');
 const { Long } = require('../../src');
@@ -1841,4 +1847,67 @@ describe('Bulk', function () {
       });
     })
   );
+
+  it('should apply collation via FindOperators', {
+    metadata: { requires: { mongodb: '>= 3.4' } },
+    test: withMonitoredClient(['update', 'delete'], function (client, events, done) {
+      const locales = ['fr', 'de', 'es'];
+      const bulk = client.db().collection('coll').initializeOrderedBulkOp();
+
+      // updates
+      bulk
+        .find({ b: 1 })
+        .collation({ locale: locales[0] })
+        .updateOne({ $set: { b: 2 } });
+      bulk
+        .find({ b: 2 })
+        .collation({ locale: locales[1] })
+        .update({ $set: { b: 3 } });
+      bulk.find({ b: 3 }).collation({ locale: locales[2] }).replaceOne({ b: 2 });
+
+      // deletes
+      bulk.find({ b: 2 }).collation({ locale: locales[0] }).removeOne();
+      bulk.find({ b: 1 }).collation({ locale: locales[1] }).remove();
+
+      bulk.execute(err => {
+        expect(err).to.not.exist;
+        expect(events).to.be.an('array').with.length.at.least(1);
+        expect(events[0]).property('commandName').to.equal('update');
+        const updateCommand = events[0].command;
+        expect(updateCommand).property('updates').to.be.an('array').with.length(3);
+        updateCommand.updates.forEach((statement, idx) => {
+          expect(statement).property('collation').to.eql({ locale: locales[idx] });
+        });
+        expect(events[1]).property('commandName').to.equal('delete');
+        const deleteCommand = events[1].command;
+        expect(deleteCommand).property('deletes').to.be.an('array').with.length(2);
+        deleteCommand.deletes.forEach((statement, idx) => {
+          expect(statement).property('collation').to.eql({ locale: locales[idx] });
+        });
+        client.close(done);
+      });
+    })
+  });
+
+  it('should throw an error if raw operations are passed to bulkWrite', function () {
+    const client = this.configuration.newClient();
+    return client.connect().then(() => {
+      this.defer(() => client.close());
+
+      const coll = client.db().collection('single_bulk_write_error');
+      return coll
+        .bulkWrite([
+          { updateOne: { q: { a: 2 }, u: { $set: { a: 2 } }, upsert: true } },
+          { deleteOne: { q: { c: 1 } } }
+        ])
+        .then(
+          () => {
+            throw new Error('expected a bulk error');
+          },
+          err => {
+            expect(err).to.match(/Raw operations are not allowed/);
+          }
+        );
+    });
+  });
 });

test/functional/collations.test.js

@@ -536,13 +536,13 @@ describe('Collation', function () {
             [
               {
                 updateOne: {
-                  q: { a: 2 },
-                  u: { $set: { a: 2 } },
+                  filter: { a: 2 },
+                  update: { $set: { a: 2 } },
                   upsert: true,
                   collation: { caseLevel: true }
                 }
               },
-              { deleteOne: { q: { c: 1 } } }
+              { deleteOne: { filter: { c: 1 } } }
             ],
             { ordered: true }
           )
@@ -559,7 +559,7 @@ describe('Collation', function () {
     }
   });
 
-  it('Successfully fail bulkWrite due to unsupported collation', {
+  it('Successfully fail bulkWrite due to unsupported collation in update', {
     metadata: { requires: { generators: true, topology: 'single' } },
     test: function () {
       const configuration = this.configuration;
@@ -588,13 +588,63 @@ describe('Collation', function () {
             [
               {
                 updateOne: {
-                  q: { a: 2 },
-                  u: { $set: { a: 2 } },
+                  filter: { a: 2 },
+                  update: { $set: { a: 2 } },
                   upsert: true,
                   collation: { caseLevel: true }
                 }
               },
-              { deleteOne: { q: { c: 1 } } }
+              { deleteOne: { filter: { c: 1 } } }
+            ],
+            { ordered: true }
+          )
+          .then(() => {
+            throw new Error('should not succeed');
+          })
+          .catch(err => {
+            expect(err).to.exist;
+            expect(err.message).to.match(/does not support collation/);
+          })
+          .then(() => client.close());
+      });
+    }
+  });
+
+  it('Successfully fail bulkWrite due to unsupported collation in delete', {
+    metadata: { requires: { generators: true, topology: 'single' } },
+    test: function () {
+      const configuration = this.configuration;
+      const client = configuration.newClient(`mongodb://${testContext.server.uri()}/test`);
+      const primary = [Object.assign({}, mock.DEFAULT_ISMASTER, { maxWireVersion: 4 })];
+
+      testContext.server.setMessageHandler(request => {
+        const doc = request.document;
+        if (doc.ismaster) {
+          request.reply(primary[0]);
+        } else if (doc.update) {
+          request.reply({ ok: 1 });
+        } else if (doc.delete) {
+          request.reply({ ok: 1 });
+        } else if (doc.endSessions) {
+          request.reply({ ok: 1 });
+        }
+      });
+
+      return client.connect().then(() => {
+        const db = client.db(configuration.db);
+
+        return db
+          .collection('test')
+          .bulkWrite(
+            [
+              {
+                updateOne: {
+                  filter: { a: 2 },
+                  update: { $set: { a: 2 } },
+                  upsert: true
+                }
+              },
+              { deleteOne: { filter: { c: 1 }, collation: { caseLevel: true } } }
             ],
             { ordered: true }
           )

test/functional/crud_api.test.js

@@ -357,12 +357,12 @@ describe('CRUD API', function () {
 
             db.collection('t2_5').bulkWrite(
               [
-                { insertOne: { a: 1 } },
+                { insertOne: { document: { a: 1 } } },
                 { insertMany: [{ g: 1 }, { g: 2 }] },
-                { updateOne: { q: { a: 2 }, u: { $set: { a: 2 } }, upsert: true } },
-                { updateMany: { q: { a: 2 }, u: { $set: { a: 2 } }, upsert: true } },
-                { deleteOne: { q: { c: 1 } } },
-                { deleteMany: { q: { c: 1 } } }
+                { updateOne: { filter: { a: 2 }, update: { $set: { a: 2 } }, upsert: true } },
+                { updateMany: { filter: { a: 2 }, update: { $set: { a: 2 } }, upsert: true } },
+                { deleteOne: { filter: { c: 1 } } },
+                { deleteMany: { filter: { c: 1 } } }
               ],
               { ordered: false, writeConcern: { w: 1 } },
               function (err, r) {
@@ -442,12 +442,12 @@ describe('CRUD API', function () {
 
             db.collection('t2_7').bulkWrite(
               [
-                { insertOne: { a: 1 } },
+                { insertOne: { document: { a: 1 } } },
                 { insertMany: [{ g: 1 }, { g: 2 }] },
-                { updateOne: { q: { a: 2 }, u: { $set: { a: 2 } }, upsert: true } },
-                { updateMany: { q: { a: 2 }, u: { $set: { a: 2 } }, upsert: true } },
-                { deleteOne: { q: { c: 1 } } },
-                { deleteMany: { q: { c: 1 } } }
+                { updateOne: { filter: { a: 2 }, update: { $set: { a: 2 } }, upsert: true } },
+                { updateMany: { filter: { a: 2 }, update: { $set: { a: 2 } }, upsert: true } },
+                { deleteOne: { filter: { c: 1 } } },
+                { deleteMany: { filter: { c: 1 } } }
               ],
               { ordered: true, writeConcern: { w: 1 } },
               function (err, r) {