Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improved Bulk Write API #1509

Draft
wants to merge 8 commits into
base: main
Choose a base branch
from
9 changes: 9 additions & 0 deletions bson/src/main/org/bson/AbstractBsonWriter.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
5 changes: 0 additions & 5 deletions bson/src/main/org/bson/BSONCallbackAdapter.java
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 0 additions & 4 deletions bson/src/main/org/bson/BsonBinaryWriter.java
Original file line number Diff line number Diff line change
Expand Up @@ -108,10 +108,6 @@ public BsonBinaryWriterSettings getBinaryWriterSettings() {
return binaryWriterSettings;
}

@Override
public void flush() {
}

@Override
protected Context getContext() {
return (Context) super.getContext();
Expand Down
4 changes: 0 additions & 4 deletions bson/src/main/org/bson/BsonDocumentWriter.java
Original file line number Diff line number Diff line change
Expand Up @@ -194,10 +194,6 @@ public void doWriteUndefined() {
write(new BsonUndefined());
}

@Override
public void flush() {
}

@Override
protected Context getContext() {
return (Context) super.getContext();
Expand Down
10 changes: 10 additions & 0 deletions bson/src/main/org/bson/io/OutputBuffer.java
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
154 changes: 154 additions & 0 deletions driver-core/src/main/com/mongodb/ClientBulkWriteException.java
Original file line number Diff line number Diff line change
@@ -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);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
/*
* 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.annotations.Sealed;
import com.mongodb.client.model.Filters;
import com.mongodb.internal.client.model.bulk.ConcreteClientBulkWriteOptions;
import com.mongodb.lang.Nullable;
import org.bson.BsonValue;
import org.bson.conversions.Bson;

/**
* The options to apply when executing a client-level bulk write operation.
*
* @since 5.3
*/
@Sealed
public interface ClientBulkWriteOptions {
/**
* Creates the default options.
*
* @return The default options.
*/
static ClientBulkWriteOptions clientBulkWriteOptions() {
return new ConcreteClientBulkWriteOptions();
}

/**
* Enables or disables ordered execution of {@linkplain ClientNamespacedWriteModel individual write operations}.
* In an ordered execution a failure of an individual operation prevents the rest of them
* from being executed.
* In an unordered execution failures of individual operations do not prevent the rest of them
* from being executed.
*
* @param ordered The ordered flag. If {@code null}, the client defaults to {@code true}.
* @return {@code this}.
*/
ClientBulkWriteOptions ordered(@Nullable Boolean ordered);

/**
* Disables or enables checking against document validation rules, a.k.a., schema validation.
*
* @param bypassDocumentValidation The flag specifying whether to bypass the document validation rules.
* {@code null} represents the server default.
* @return {@code this}.
*/
ClientBulkWriteOptions bypassDocumentValidation(@Nullable Boolean bypassDocumentValidation);

/**
* Sets variables that can be referenced from {@linkplain ClientNamespacedWriteModel individual write operations}
* with the {@code "$$"} syntax, which in turn requires using {@link Filters#expr(Object)} when specifying filters.
* Values must be constants or expressions that do not reference fields.
*
* @param let The variables. {@code null} represents the server default.
* @return {@code this}.
* @mongodb.driver.manual reference/aggregation-variables/ Variables in Aggregation Expressions
*/
ClientBulkWriteOptions let(@Nullable Bson let);

/**
* Sets the comment to attach to the {@code bulkWrite} administration command.
*
* @param comment The comment. {@code null} represents the server default.
* @return {@code this}.
*/
ClientBulkWriteOptions comment(@Nullable BsonValue comment);

/**
* Enables or disables requesting {@linkplain ClientBulkWriteResult#getVerboseResults() verbose results}.
*
* @param verboseResults The flag specifying whether to request verbose results.
* If {@code null}, the client defaults to {@code false}.
* This value corresponds inversely to the {@code errorsOnly} field of the {@code bulkWrite} administration command.
* @return {@code this}.
*/
ClientBulkWriteOptions verboseResults(@Nullable Boolean verboseResults);
}
Loading