Introduce Client Bulk Write API (#1509)

Introduce the Client Bulk Write API to support write operations across multiple collections and databases. This update allows mixed operations and cross-namespace bulk writes to be executed in a single batch: https://www.mongodb.com/docs/manual/reference/command/bulkWrite/.

JAVA-4586
---------

Co-authored-by: Valentin Kovalenko <valentin.kovalenko@gmail.com>
Co-authored-by: Viacheslav Babanin <slav.babanin@mongodb.com>
Co-authored-by: Jeff Yemin <jeff.yemin@mongodb.com>

bson/src/main/org/bson/AbstractBsonWriter.java

@@ -748,6 +748,15 @@ protected void throwInvalidState(final String methodName, final State... validSt
                                                        methodName, validStatesString, state));
     }
 
+    /**
+     * {@inheritDoc}
+     * <p>
+     * The {@link #flush()} method of {@link AbstractBsonWriter} does nothing.</p>
+     */
+    @Override
+    public void flush() {
+    }
+
     @Override
     public void close() {
         closed = true;

bson/src/main/org/bson/BSONCallbackAdapter.java

@@ -39,11 +39,6 @@ protected BSONCallbackAdapter(final BsonWriterSettings settings, final BSONCallb
         this.bsonCallback = bsonCallback;
     }
 
-    @Override
-    public void flush() {
-        //Looks like should be no-op?
-    }
-
     @Override
     public void doWriteStartDocument() {
         BsonContextType contextType = getState() == State.SCOPE_DOCUMENT

bson/src/main/org/bson/BsonBinaryWriter.java

@@ -108,10 +108,6 @@ public BsonBinaryWriterSettings getBinaryWriterSettings() {
         return binaryWriterSettings;
     }
 
-    @Override
-    public void flush() {
-    }
-
     @Override
     protected Context getContext() {
         return (Context) super.getContext();

bson/src/main/org/bson/BsonDocumentWriter.java

@@ -194,10 +194,6 @@ public void doWriteUndefined() {
         write(new BsonUndefined());
     }
 
-    @Override
-    public void flush() {
-    }
-
     @Override
     protected Context getContext() {
         return (Context) super.getContext();

bson/src/main/org/bson/io/OutputBuffer.java

@@ -41,6 +41,16 @@ public void write(final byte[] b) {
     public void close() {
     }
 
+    /**
+     * {@inheritDoc}
+     * <p>
+     * The {@link #flush()} method of {@link OutputBuffer} does nothing.</p>
+     */
+    @Override
+    public void flush() throws IOException {
+        super.flush();
+    }
+
     @Override
     public void write(final byte[] bytes, final int offset, final int length) {
         writeBytes(bytes, offset, length);

build.gradle

@@ -59,6 +59,7 @@ ext {
     junitBomVersion = '5.10.2'
     logbackVersion = '1.3.14'
     graalSdkVersion = '24.0.0'
+    reflectionsVersion = '0.9.10'
     gitVersion = getGitVersion()
 }
 
@@ -128,7 +129,7 @@ configure(scalaProjects) {
         testImplementation('org.scalatestplus:junit-4-13_%%:3.2.9.0')
         testImplementation('org.scalatestplus:mockito-3-12_%%:3.2.10.0')
         testImplementation("ch.qos.logback:logback-classic:$logbackVersion")
-        testImplementation('org.reflections:reflections:0.9.10')
+        testImplementation("org.reflections:reflections:$reflectionsVersion")
     }
 
     test{

config/detekt/detekt.yml

@@ -159,7 +159,7 @@ complexity:
     active: true
     excludes: ['**/test/**']
     thresholdInFiles: 25
-    thresholdInClasses: 25
+    thresholdInClasses: 27
     thresholdInInterfaces: 25
     thresholdInObjects: 25
     thresholdInEnums: 25

driver-core/build.gradle

@@ -60,6 +60,7 @@ dependencies {
 
     testImplementation project(':bson').sourceSets.test.output
     testImplementation('org.junit.jupiter:junit-jupiter-api')
+    testImplementation("org.reflections:reflections:$reflectionsVersion")
     testRuntimeOnly "io.netty:netty-tcnative-boringssl-static"
 
     classifiers.forEach {

driver-core/src/main/com/mongodb/ClientBulkWriteException.java

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb;
+
+import com.mongodb.bulk.WriteConcernError;
+import com.mongodb.client.model.bulk.ClientBulkWriteResult;
+import com.mongodb.client.model.bulk.ClientNamespacedWriteModel;
+import com.mongodb.lang.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.mongodb.assertions.Assertions.isTrueArgument;
+import static com.mongodb.assertions.Assertions.notNull;
+import static com.mongodb.internal.operation.ClientBulkWriteOperation.Exceptions.serverAddressFromException;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.unmodifiableList;
+import static java.util.Collections.unmodifiableMap;
+import static java.util.Optional.ofNullable;
+
+/**
+ * The result of an unsuccessful or partially unsuccessful client-level bulk write operation.
+ * Note that the {@linkplain #getCode() code} and {@linkplain #getErrorLabels() labels} from this exception are not useful.
+ * An application should use those from the {@linkplain #getCause() top-level error}.
+ *
+ * @see ClientBulkWriteResult
+ * @since 5.3
+ * @serial exclude
+ */
+public final class ClientBulkWriteException extends MongoServerException {
+    private static final long serialVersionUID = 1;
+
+    private final List<WriteConcernError> writeConcernErrors;
+    private final Map<Integer, WriteError> writeErrors;
+    @Nullable
+    private final ClientBulkWriteResult partialResult;
+
+    /**
+     * Constructs a new instance.
+     *
+     * @param error The {@linkplain #getCause() top-level error}.
+     * @param writeConcernErrors The {@linkplain #getWriteConcernErrors() write concern errors}.
+     * @param writeErrors The {@linkplain #getWriteErrors() write errors}.
+     * @param partialResult The {@linkplain #getPartialResult() partial result}.
+     * @param serverAddress The {@linkplain MongoServerException#getServerAddress() server address}.
+     * If {@code error} is a {@link MongoServerException} or a {@link MongoSocketException}, then {@code serverAddress}
+     * must be equal to the {@link ServerAddress} they bear.
+     */
+    public ClientBulkWriteException(
+            @Nullable final MongoException error,
+            @Nullable final List<WriteConcernError> writeConcernErrors,
+            @Nullable final Map<Integer, WriteError> writeErrors,
+            @Nullable final ClientBulkWriteResult partialResult,
+            final ServerAddress serverAddress) {
+        super(
+                message(
+                        error, writeConcernErrors, writeErrors, partialResult,
+                        notNull("serverAddress", serverAddress)),
+                validateServerAddress(error, serverAddress));
+        initCause(error);
+        isTrueArgument("At least one of `writeConcernErrors`, `writeErrors`, `partialResult` must be non-null or non-empty",
+                !(writeConcernErrors == null || writeConcernErrors.isEmpty())
+                        || !(writeErrors == null || writeErrors.isEmpty())
+                        || partialResult != null);
+        this.writeConcernErrors = writeConcernErrors == null ? emptyList() : unmodifiableList(writeConcernErrors);
+        this.writeErrors = writeErrors == null ? emptyMap() : unmodifiableMap(writeErrors);
+        this.partialResult = partialResult;
+    }
+
+    private static String message(
+            @Nullable final MongoException error,
+            @Nullable final List<WriteConcernError> writeConcernErrors,
+            @Nullable final Map<Integer, WriteError> writeErrors,
+            @Nullable final ClientBulkWriteResult partialResult,
+            final ServerAddress serverAddress) {
+        return "Client-level bulk write operation error on server " + serverAddress + "."
+                + (error == null ? "" : " Top-level error: " + error + ".")
+                + (writeErrors == null || writeErrors.isEmpty() ? "" : " Write errors: " + writeErrors + ".")
+                + (writeConcernErrors == null || writeConcernErrors.isEmpty() ? "" : " Write concern errors: " + writeConcernErrors + ".")
+                + (partialResult == null ? "" : " Partial result: " + partialResult + ".");
+    }
+
+    private static ServerAddress validateServerAddress(@Nullable final MongoException error, final ServerAddress serverAddress) {
+        serverAddressFromException(error).ifPresent(serverAddressFromError ->
+                isTrueArgument("`serverAddress` must be equal to that of the `error`", serverAddressFromError.equals(serverAddress)));
+        return error instanceof MongoServerException
+                ? ((MongoServerException) error).getServerAddress()
+                : serverAddress;
+    }
+
+    /**
+     * The top-level error. That is an error that is neither a {@linkplain #getWriteConcernErrors() write concern error},
+     * nor is an {@linkplain #getWriteErrors() error of an individual write operation}.
+     *
+     * @return The top-level error. Non-{@code null} only if a top-level error occurred.
+     */
+    @Override
+    @Nullable
+    public MongoException getCause() {
+        return (MongoException) super.getCause();
+    }
+
+    /**
+     * The {@link WriteConcernError}s that occurred while executing the client-level bulk write operation.
+     * <p>
+     * There are no guarantees on mutability of the {@link List} returned.</p>
+     *
+     * @return The {@link WriteConcernError}s.
+     */
+    public List<WriteConcernError> getWriteConcernErrors() {
+        return writeConcernErrors;
+    }
+
+    /**
+     * The indexed {@link WriteError}s.
+     * The {@linkplain Map#keySet() keys} are the indexes of the corresponding {@link ClientNamespacedWriteModel}s
+     * in the corresponding client-level bulk write operation.
+     * <p>
+     * There are no guarantees on mutability or iteration order of the {@link Map} returned.</p>
+     *
+     * @return The indexed {@link WriteError}s.
+     * @see ClientBulkWriteResult.VerboseResults#getInsertResults()
+     * @see ClientBulkWriteResult.VerboseResults#getUpdateResults()
+     * @see ClientBulkWriteResult.VerboseResults#getDeleteResults()
+     */
+    public Map<Integer, WriteError> getWriteErrors() {
+        return writeErrors;
+    }
+
+    /**
+     * The result of the part of a client-level bulk write operation that is known to be successful.
+     *
+     * @return The successful partial result. {@linkplain Optional#isPresent() Present} only if the client received a response indicating success
+     * of at least one {@linkplain ClientNamespacedWriteModel individual write operation}.
+     */
+    public Optional<ClientBulkWriteResult> getPartialResult() {
+        return ofNullable(partialResult);
+    }
+}

driver-core/src/main/com/mongodb/client/model/bulk/BaseClientDeleteOptions.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.client.model.bulk;
+
+/**
+ * The methods declared in this interface are part of the public API of subclasses or sub-interfaces.
+ */
+interface BaseClientDeleteOptions extends BaseClientWriteModelOptions {
+}

driver-core/src/main/com/mongodb/client/model/bulk/BaseClientUpdateOptions.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.client.model.bulk;
+
+import com.mongodb.lang.Nullable;
+import org.bson.conversions.Bson;
+
+/**
+ * The methods declared in this interface are part of the public API of subclasses or sub-interfaces.
+ */
+interface BaseClientUpdateOptions extends BaseClientWriteModelOptions, BaseClientUpsertableWriteModelOptions {
+
+    BaseClientUpdateOptions arrayFilters(@Nullable Iterable<? extends Bson> arrayFilters);
+}

driver-core/src/main/com/mongodb/client/model/bulk/BaseClientUpsertableWriteModelOptions.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.client.model.bulk;
+
+import com.mongodb.lang.Nullable;
+
+/**
+ * The methods declared in this interface are part of the public API of subclasses or sub-interfaces.
+ */
+interface BaseClientUpsertableWriteModelOptions {
+    BaseClientUpsertableWriteModelOptions upsert(@Nullable Boolean upsert);
+}

driver-core/src/main/com/mongodb/client/model/bulk/BaseClientWriteModelOptions.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.client.model.bulk;
+
+import com.mongodb.client.model.Collation;
+import com.mongodb.lang.Nullable;
+import org.bson.conversions.Bson;
+
+/**
+ * The methods declared in this interface are part of the public API of subclasses or sub-interfaces.
+ */
+interface BaseClientWriteModelOptions {
+    BaseClientWriteModelOptions collation(@Nullable Collation collation);
+
+    BaseClientWriteModelOptions hint(@Nullable Bson hint);
+
+    BaseClientWriteModelOptions hintString(@Nullable String hintString);
+}