summaryrefslogtreecommitdiff
path: root/src/mongo/s/write_ops
diff options
context:
space:
mode:
authorLucas de Castro Borges <lucas@gnuabordo.com.br>2025-02-11 15:07:35 -0300
committerLucas de Castro Borges <lucas@gnuabordo.com.br>2025-02-11 15:07:35 -0300
commit4cb8841196d0625dfa3825aa326f071cd27c7b8b (patch)
tree1682a647d4463397c119183369ae6f750d5fdcff /src/mongo/s/write_ops
parentaa03c6362cbaa767638e6eed9b031d86dd2643d1 (diff)
parent8f0827553e09872941945a093b647a4211a9db7f (diff)
Update upstream source from tag 'upstream/6.0.0'master
Update to upstream version '6.0.0' with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322
Diffstat (limited to 'src/mongo/s/write_ops')
-rw-r--r--src/mongo/s/write_ops/batch_write_exec.cpp24
-rw-r--r--src/mongo/s/write_ops/batch_write_exec_test.cpp508
-rw-r--r--src/mongo/s/write_ops/batch_write_op.cpp78
-rw-r--r--src/mongo/s/write_ops/batch_write_op_test.cpp29
-rw-r--r--src/mongo/s/write_ops/batched_command_request.cpp27
-rw-r--r--src/mongo/s/write_ops/batched_command_request.h34
-rw-r--r--src/mongo/s/write_ops/batched_command_request_test.cpp2
-rw-r--r--src/mongo/s/write_ops/write_op.cpp51
8 files changed, 162 insertions, 591 deletions
diff --git a/src/mongo/s/write_ops/batch_write_exec.cpp b/src/mongo/s/write_ops/batch_write_exec.cpp
index 980d376b607..45b89579dbb 100644
--- a/src/mongo/s/write_ops/batch_write_exec.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec.cpp
@@ -104,25 +104,6 @@ bool hasTransientTransactionError(const BatchedCommandResponse& response) {
// applies when no writes are occurring and metadata is not changing on reload.
const int kMaxRoundsWithoutProgress(5);
-/**
- * Provides the write concern with which child batches have to be internally submitted.
- */
-boost::optional<WriteConcernOptions> getWriteConcernForChildBatch(OperationContext* opCtx) {
- // Per-operation write concern is not supported in transactions.
- if (TransactionRouter::get(opCtx)) {
- return boost::none;
- }
-
- // Retrieve the WC specified by the remote client; in case of "fire and forget" request, the WC
- // needs to be upgraded to "w: 1" for the sharding protocol to correctly handle internal
- // writeErrors.
- auto wc = opCtx->getWriteConcern();
- if (!wc.requiresWriteAcknowledgement()) {
- wc.w = 1;
- }
-
- return wc;
-}
} // namespace
void BatchWriteExec::executeBatch(OperationContext* opCtx,
@@ -214,7 +195,6 @@ void BatchWriteExec::executeBatch(OperationContext* opCtx,
//
std::vector<AsyncRequestsSender::Request> requests;
- const auto wcSettingForChildBatch = getWriteConcernForChildBatch(opCtx);
// Get as many batches as we can at once
for (auto&& childBatch : childBatches) {
@@ -237,10 +217,6 @@ void BatchWriteExec::executeBatch(OperationContext* opCtx,
BSONObjBuilder requestBuilder;
shardBatchRequest.serialize(&requestBuilder);
logical_session_id_helpers::serializeLsidAndTxnNumber(opCtx, &requestBuilder);
- if (wcSettingForChildBatch) {
- requestBuilder.append(WriteConcernOptions::kWriteConcernField,
- wcSettingForChildBatch->toBSON());
- }
return requestBuilder.obj();
}();
diff --git a/src/mongo/s/write_ops/batch_write_exec_test.cpp b/src/mongo/s/write_ops/batch_write_exec_test.cpp
index 3e90e5b65d3..3b4079b4c84 100644
--- a/src/mongo/s/write_ops/batch_write_exec_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec_test.cpp
@@ -33,14 +33,12 @@
#include "mongo/client/remote_command_targeter_factory_mock.h"
#include "mongo/client/remote_command_targeter_mock.h"
#include "mongo/db/commands.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
#include "mongo/db/logical_session_id.h"
#include "mongo/db/vector_clock.h"
#include "mongo/s/catalog/type_shard.h"
#include "mongo/s/client/shard_registry.h"
#include "mongo/s/mock_ns_targeter.h"
#include "mongo/s/session_catalog_router.h"
-#include "mongo/s/shard_cannot_refresh_due_to_locks_held_exception.h"
#include "mongo/s/sharding_router_test_fixture.h"
#include "mongo/s/stale_exception.h"
#include "mongo/s/transaction_router.h"
@@ -202,40 +200,6 @@ BSONObj expectInsertsReturnTenantMigrationAbortedErrorsBase(
return tenantMigrationAbortedResponse.obj();
}
-BSONObj expectInsertsReturnCannotRefreshErrorsBase(const NamespaceString& nss,
- const std::vector<BSONObj>& expected,
- const executor::RemoteCommandRequest& request) {
- ASSERT_EQUALS(nss.db(), request.dbname);
-
- const auto opMsgRequest(OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj));
- const auto actualBatchedInsert(BatchedCommandRequest::parseInsert(opMsgRequest));
- ASSERT_EQUALS(nss.toString(), actualBatchedInsert.getNS().ns());
-
- const auto& inserted = actualBatchedInsert.getInsertRequest().getDocuments();
- ASSERT_EQUALS(expected.size(), inserted.size());
-
- auto itInserted = inserted.begin();
- auto itExpected = expected.begin();
-
- for (; itInserted != inserted.end(); itInserted++, itExpected++) {
- ASSERT_BSONOBJ_EQ(*itExpected, *itInserted);
- }
-
- BatchedCommandResponse cannotRefreshResponse;
- cannotRefreshResponse.setStatus(Status::OK());
- cannotRefreshResponse.setN(0);
-
- // Report a ShardCannotRefreshDueToLocksHeld error for each write in the batch.
- int i = 0;
- for (itInserted = inserted.begin(); itInserted != inserted.end(); ++itInserted) {
- cannotRefreshResponse.addToErrDetails(write_ops::WriteError(
- i, Status(ShardCannotRefreshDueToLocksHeldInfo(nss), "Catalog cache busy in refresh")));
- ++i;
- }
-
- return cannotRefreshResponse.toBSON();
-}
-
/**
* Mimics a single shard backend for a particular collection which can be initialized with a
* set of write command results to return.
@@ -335,12 +299,6 @@ public:
});
}
- void expectInsertsReturnCannotRefreshErrors(const std::vector<BSONObj>& expected) {
- onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) {
- return expectInsertsReturnCannotRefreshErrorsBase(nss, expected, request);
- });
- }
-
void expectInsertsReturnError(const std::vector<BSONObj>& expected,
const BatchedCommandResponse& errResponse) {
onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) {
@@ -397,6 +355,7 @@ TEST_F(BatchWriteExecTest, SingleOpUnordered) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
// Do single-target, single doc batch write op
auto future = launchAsync([&] {
@@ -435,6 +394,7 @@ TEST_F(BatchWriteExecTest, SingleUpdateTargetsShardWithLet) {
<< "100")))});
return updateOp;
}());
+ updateRequest.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -520,6 +480,7 @@ TEST_F(BatchWriteExecTest, SingleDeleteTargetsShardWithLet) {
deleteOp.setDeletes(std::vector{write_ops::DeleteOpEntry(q, false)});
return deleteOp;
}());
+ deleteRequest.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
@@ -605,6 +566,7 @@ TEST_F(BatchWriteExecTest, MultiOpLargeOrdered) {
insertOp.setDocuments(docsToInsert);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -637,6 +599,7 @@ TEST_F(BatchWriteExecTest, SingleOpUnorderedError) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -677,6 +640,7 @@ TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithStaleShardVersionError) {
insertOp.setDocuments(docsToInsert);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -708,6 +672,7 @@ TEST_F(BatchWriteExecTest, StaleShardVersionReturnedFromBatchWithSingleMultiWrit
write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -785,43 +750,6 @@ TEST_F(BatchWriteExecTest, StaleShardVersionReturnedFromBatchWithSingleMultiWrit
ASSERT_EQ(3, response.getNModified());
}
-TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithCannotRefreshError) {
- const int kNumDocsToInsert = 100'000;
-
- std::vector<BSONObj> docsToInsert;
- docsToInsert.reserve(kNumDocsToInsert);
- for (int i = 0; i < kNumDocsToInsert; i++) {
- docsToInsert.push_back(BSON("_id" << i));
- }
-
- BatchedCommandRequest request([&] {
- write_ops::InsertCommandRequest insertOp(nss);
- insertOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(false);
- return writeCommandBase;
- }());
- insertOp.setDocuments(docsToInsert);
- return insertOp;
- }());
-
- auto future = launchAsync([&] {
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
- BatchWriteExec::executeBatch(
- operationContext(), singleShardNSTargeter, request, &response, &stats);
-
- ASSERT(response.getOk());
- ASSERT_EQ(kNumDocsToInsert, response.getN());
- });
-
- expectInsertsReturnCannotRefreshErrors({docsToInsert.begin(), docsToInsert.begin() + 60133});
- expectInsertsReturnSuccess({docsToInsert.begin(), docsToInsert.begin() + 60133});
- expectInsertsReturnSuccess({docsToInsert.begin() + 60133, docsToInsert.end()});
-
- future.default_timed_get();
-}
-
TEST_F(BatchWriteExecTest,
RetryableErrorReturnedFromMultiWriteWithShard1AllOKShard2AllStaleShardVersion) {
BatchedCommandRequest request([&] {
@@ -840,6 +768,7 @@ TEST_F(BatchWriteExecTest,
write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -943,6 +872,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromMultiWriteWithShard1Firs) {
write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -1056,6 +986,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromMultiWriteWithShard1FirstOK
write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -1165,6 +1096,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromWriteWithShard1SSVShard2OK)
write_ops::UpdateModification::parseFromClassicUpdate(BSON("x" << 1)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -1264,6 +1196,7 @@ TEST_F(BatchWriteExecTest, StaleShardOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -1295,6 +1228,7 @@ TEST_F(BatchWriteExecTest, MultiStaleShardOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1332,6 +1266,7 @@ TEST_F(BatchWriteExecTest, TooManyStaleShardOp) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1366,6 +1301,7 @@ TEST_F(BatchWriteExecTest, StaleDbOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -1397,6 +1333,7 @@ TEST_F(BatchWriteExecTest, MultiStaleDbOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1434,6 +1371,7 @@ TEST_F(BatchWriteExecTest, TooManyStaleDbOp) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1457,73 +1395,6 @@ TEST_F(BatchWriteExecTest, TooManyStaleDbOp) {
future.default_timed_get();
}
-TEST_F(BatchWriteExecTest, MultiCannotRefreshShardOp) {
- BatchedCommandRequest request([&] {
- write_ops::InsertCommandRequest insertOp(nss);
- insertOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(false);
- return writeCommandBase;
- }());
- insertOp.setDocuments({BSON("x" << 1)});
- return insertOp;
- }());
-
- auto future = launchAsync([&] {
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
- BatchWriteExec::executeBatch(
- operationContext(), singleShardNSTargeter, request, &response, &stats);
- ASSERT(response.getOk());
- });
-
- const std::vector<BSONObj> expected{BSON("x" << 1)};
-
- // Return multiple ShardCannotRefreshDueToLocksHeld errors, but less than the give-up number
- for (int i = 0; i < 3; i++) {
- expectInsertsReturnCannotRefreshErrors(expected);
- }
-
- expectInsertsReturnSuccess(expected);
-
- future.default_timed_get();
-}
-
-TEST_F(BatchWriteExecTest, TooManyCannotRefreshShardOp) {
- // Retry op in exec too many times b/c of busy catalog cache (the error is not expected to
- // trigger a refresh on any implementation of NSTargeter). We should report a no progress error
- // for everything in the batch.
- BatchedCommandRequest request([&] {
- write_ops::InsertCommandRequest insertOp(nss);
- insertOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(false);
- return writeCommandBase;
- }());
- insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
- return insertOp;
- }());
-
- auto future = launchAsync([&] {
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
- BatchWriteExec::executeBatch(
- operationContext(), singleShardNSTargeter, request, &response, &stats);
- ASSERT(response.getOk());
- ASSERT_EQ(0, response.getN());
- ASSERT(response.isErrDetailsSet());
- ASSERT_EQUALS(response.getErrDetailsAt(0).getStatus().code(), ErrorCodes::NoProgressMade);
- ASSERT_EQUALS(response.getErrDetailsAt(1).getStatus().code(), ErrorCodes::NoProgressMade);
- });
-
- // Return multiple StaleShardVersion errors
- for (int i = 0; i < (1 + kMaxRoundsWithoutProgress); i++) {
- expectInsertsReturnCannotRefreshErrors({BSON("x" << 1), BSON("x" << 2)});
- }
-
- future.default_timed_get();
-}
-
TEST_F(BatchWriteExecTest, RetryableWritesLargeBatch) {
// A retryable error without a txnNumber is not retried.
@@ -1546,6 +1417,7 @@ TEST_F(BatchWriteExecTest, RetryableWritesLargeBatch) {
insertOp.setDocuments(docsToInsert);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1580,6 +1452,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorNoTxnNumber) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
BatchedCommandResponse retryableErrResponse;
retryableErrResponse.setStatus({ErrorCodes::NotWritablePrimary, "mock retryable error"});
@@ -1618,6 +1491,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorTxnNumber) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1655,6 +1529,7 @@ TEST_F(BatchWriteExecTest, NonRetryableErrorTxnNumber) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1696,6 +1571,7 @@ TEST_F(BatchWriteExecTest, StaleEpochIsNotRetryable) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1723,249 +1599,6 @@ TEST_F(BatchWriteExecTest, StaleEpochIsNotRetryable) {
future.default_timed_get();
}
-TEST_F(BatchWriteExecTest, FireAndForgetBatchInsertGetsReplyWithOnlyOkStatus) {
- const int kNumDocsToInsert = 5;
- const std::string kDocValue("sample");
-
- std::vector<BSONObj> docsToInsert;
- docsToInsert.reserve(kNumDocsToInsert);
- for (int i = 0; i < kNumDocsToInsert; i++) {
- docsToInsert.push_back(BSON("_id" << i << "otherField" << kDocValue));
- }
-
- BatchedCommandRequest request([&] {
- write_ops::InsertCommandRequest insertOp(nss);
- insertOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(true);
- return writeCommandBase;
- }());
- insertOp.setDocuments(docsToInsert);
- return insertOp;
- }());
-
- auto future = launchAsync([&] {
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
- // Set Unacknowledged WC for a "fire & forget" request
- auto opCtx = operationContext();
- opCtx->setWriteConcern(
- WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue());
- BatchWriteExec::executeBatch(opCtx, singleShardNSTargeter, request, &response, &stats);
-
- // The reply should only contain an OK status, without any further detail on
- // the ops actually executed across the cluster.
- BatchedCommandResponse expectedReplyToFireAndForgetRequest;
- expectedReplyToFireAndForgetRequest.setStatus(Status::OK());
- ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0);
- });
-
- expectInsertsReturnSuccess(docsToInsert.begin(), docsToInsert.end());
-
- future.default_timed_get();
-}
-
-TEST_F(BatchWriteExecTest, FireAndForgetBatchUpdateGetsReplyWithOnlyOkStatus) {
- BatchedCommandRequest request([&] {
- write_ops::UpdateCommandRequest updateOp(nss);
- updateOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(false);
- return writeCommandBase;
- }());
- updateOp.setUpdates(std::vector{write_ops::UpdateOpEntry(
- BSON("_id" << 100),
- write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
- return updateOp;
- }());
-
- const static auto epoch = OID::gen();
- const static Timestamp timestamp(2);
-
- // This allows the batch to target each write operation to perform this test
- class MultiShardTargeter : public MockNSTargeter {
- public:
- using MockNSTargeter::MockNSTargeter;
-
- std::vector<ShardEndpoint> targetUpdate(OperationContext* opCtx,
- const BatchItemRef& itemRef) const override {
- if (targetAll) {
- return std::vector{
- ShardEndpoint(
- kShardName1, ChunkVersion(100, 200, epoch, timestamp), boost::none),
- ShardEndpoint(
- kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none)};
- } else {
- return std::vector{ShardEndpoint(
- kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none)};
- }
- }
-
- bool targetAll = true;
- };
-
- MultiShardTargeter multiShardNSTargeter(
- nss,
- {MockRange(
- ShardEndpoint(kShardName1, ChunkVersion(100, 200, epoch, timestamp), boost::none),
- BSON("sk" << MINKEY),
- BSON("sk" << 10)),
- MockRange(
- ShardEndpoint(kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none),
- BSON("sk" << 10),
- BSON("sk" << MAXKEY))});
- auto future = launchAsync([&] {
- // Set Unacknowledged WC for a "fire & forget" request
- auto opCtx = operationContext();
- opCtx->setWriteConcern(
- WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue());
-
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
- BatchWriteExec::executeBatch(opCtx, multiShardNSTargeter, request, &response, &stats);
- return response;
- });
-
- onCommandForPoolExecutor([&](const RemoteCommandRequest& request) {
- ASSERT_EQ(kTestShardHost1, request.target);
-
- BatchedCommandResponse response;
- response.setStatus(Status::OK());
- response.setNModified(1);
-
- return response.toBSON();
- });
-
- onCommandForPoolExecutor([&](const RemoteCommandRequest& request) {
- ASSERT_EQ(kTestShardHost2, request.target);
-
- BatchedCommandResponse response;
- response.setStatus(Status::OK());
- response.setNModified(0);
- response.addToErrDetails(
- write_ops::WriteError(0,
- Status(StaleConfigInfo(nss,
- ChunkVersion(101, 200, epoch, timestamp),
- ChunkVersion(105, 200, epoch, timestamp),
- ShardId(kShardName2)),
- "Stale error")));
- return response.toBSON();
- });
-
- onCommandForPoolExecutor([&](const RemoteCommandRequest& request) {
- ASSERT_EQ(kTestShardHost2, request.target);
-
- BatchedCommandResponse response;
- response.setStatus(Status::OK());
- response.setNModified(2);
-
- return response.toBSON();
- });
-
- // The reply should only contain an OK status, without any further detail on
- // the ops actually executed across the cluster.
- // (Despite of a "fire and forget" request, child batches still need to be internally processed
- // before returning a reply).
- auto response = future.default_timed_get();
- BatchedCommandResponse expectedReplyToFireAndForgetRequest;
- expectedReplyToFireAndForgetRequest.setStatus(Status::OK());
- ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0);
-}
-
-TEST_F(BatchWriteExecTest, FireAndForgetBatchDeleteGetsReplyWithOnlyOkStatus) {
- // Try to update the single doc where a let param is used in the shard key.
- const auto let = BSON("y" << 100);
- const auto rtc = LegacyRuntimeConstants{Date_t::now(), Timestamp(1, 1)};
- const auto q = BSON("x"
- << "$$y");
- BatchedCommandRequest deleteRequest([&] {
- write_ops::DeleteCommandRequest deleteOp(nss);
- deleteOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(false);
- return writeCommandBase;
- }());
- deleteOp.setLet(let);
- deleteOp.setLegacyRuntimeConstants(rtc);
- deleteOp.setDeletes(std::vector{write_ops::DeleteOpEntry(q, false)});
- return deleteOp;
- }());
-
- const static auto epoch = OID::gen();
-
- class MultiShardTargeter : public MockNSTargeter {
- public:
- using MockNSTargeter::MockNSTargeter;
-
- protected:
- std::vector<ShardEndpoint> targetDelete(OperationContext* opCtx,
- const BatchItemRef& itemRef) const override {
- return std::vector{ShardEndpoint(
- kShardName2, ChunkVersion(101, 200, epoch, Timestamp(1, 1)), boost::none)};
- }
- };
-
- MultiShardTargeter multiShardNSTargeter(
- nss,
- {MockRange(ShardEndpoint(
- kShardName1, ChunkVersion(100, 200, epoch, Timestamp(1, 1)), boost::none),
- BSON("x" << MINKEY),
- BSON("x" << 0)),
- MockRange(ShardEndpoint(
- kShardName2, ChunkVersion(101, 200, epoch, Timestamp(1, 1)), boost::none),
- BSON("x" << 0),
- BSON("x" << MAXKEY))});
-
- auto future = launchAsync([&] {
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
-
- // Set Unacknowledged WC for a "fire & forget" request
- auto opCtx = operationContext();
- opCtx->setWriteConcern(
- WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue());
-
- BatchWriteExec::executeBatch(opCtx, multiShardNSTargeter, deleteRequest, &response, &stats);
-
- return response;
- });
-
- // The update will hit the first shard.
- onCommandForPoolExecutor(
- [&](const RemoteCommandRequest& request) {
- ASSERT_EQ(kTestShardHost2, request.target);
-
- BatchedCommandResponse response;
- response.setStatus(Status::OK());
-
- // Check that let params are propagated to shards.
- const auto opMsgRequest(OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj));
- const auto actualBatchedUpdate(BatchedCommandRequest::parseDelete(opMsgRequest));
- ASSERT_BSONOBJ_EQ(let, actualBatchedUpdate.getLet().value_or(BSONObj()));
- ASSERT_EQUALS(actualBatchedUpdate.getLegacyRuntimeConstants()->getLocalNow(),
- rtc.getLocalNow());
- ASSERT_EQUALS(actualBatchedUpdate.getLegacyRuntimeConstants()->getClusterTime(),
- rtc.getClusterTime());
-
- // Check that let params are only forwarded and not evaluated.
- auto expectedQ = BSON("x"
- << "$$y");
- for (auto&& u : actualBatchedUpdate.getDeleteRequest().getDeletes())
- ASSERT_BSONOBJ_EQ(expectedQ, u.getQ());
-
- return response.toBSON();
- });
-
- // The reply should only contain an OK status, without any further detail on
- // the ops actually executed across the cluster.
- // (Despite of a "fire and forget" request, child batches still need to be internally processed
- // before returning a reply).
- auto response = future.default_timed_get();
- BatchedCommandResponse expectedReplyToFireAndForgetRequest;
- expectedReplyToFireAndForgetRequest.setStatus(Status::OK());
- ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0);
-}
-
TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorOrderedOp) {
const std::vector<BSONObj> expected{BSON("x" << 1), BSON("x" << 2), BSON("x" << 3)};
BatchedCommandRequest request([&] {
@@ -1978,6 +1611,7 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorOrderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -2008,6 +1642,7 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorUnorderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -2038,6 +1673,7 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorUnorderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
const int numTenantMigrationAbortedErrors = 3;
@@ -2072,6 +1708,7 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorOrderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
const int numTenantMigrationAbortedErrors = 3;
@@ -2106,6 +1743,7 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationAbortedErrorOrderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -2138,6 +1776,7 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationErrorUnorderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -2222,6 +1861,7 @@ TEST_F(BatchWriteExecTargeterErrorTest, TargetedFailedAndErrorResponse) {
write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -2357,6 +1997,7 @@ TEST_F(BatchWriteExecTransactionTargeterErrorTest, TargetedFailedAndErrorRespons
write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -2500,6 +2141,7 @@ TEST_F(BatchWriteExecTransactionMultiShardTest, TargetedSucceededAndErrorRespons
write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))});
return updateOp;
}());
+ request.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -2625,22 +2267,6 @@ public:
});
}
- void expectInsertsReturnCannotRefreshErrors(const std::vector<BSONObj>& expected) {
- onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) {
- BSONObjBuilder bob;
-
- bob.appendElementsUnique(
- expectInsertsReturnCannotRefreshErrorsBase(nss, expected, request));
-
- // Because this is the transaction-specific fixture, return transaction metadata in
- // the response.
- TxnResponseMetadata txnResponseMetadata(false /* readOnly */);
- txnResponseMetadata.serialize(&bob);
-
- return bob.obj();
- });
- }
-
void expectInsertsReturnTransientTxnErrors(const std::vector<BSONObj>& expected) {
onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) {
ASSERT_EQUALS(nss.db(), request.dbname);
@@ -2690,6 +2316,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchThrows_CommandError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2721,6 +2348,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2750,6 +2378,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrdered) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2768,64 +2397,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrdered) {
future.default_timed_get();
}
-TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorFromBusyCache) {
- BatchedCommandRequest request([&] {
- write_ops::InsertCommandRequest insertOp(nss);
- insertOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(false);
- return writeCommandBase;
- }());
- insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
- return insertOp;
- }());
-
- auto future = launchAsync([&] {
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
- BatchWriteExec::executeBatch(
- operationContext(), singleShardNSTargeter, request, &response, &stats);
-
- ASSERT(response.isErrDetailsSet());
- ASSERT_GT(response.sizeErrDetails(), 0u);
- ASSERT_EQ(ErrorCodes::ShardCannotRefreshDueToLocksHeld,
- response.getErrDetailsAt(0).getStatus().code());
- });
-
- expectInsertsReturnCannotRefreshErrors({BSON("x" << 1), BSON("x" << 2)});
-
- future.default_timed_get();
-}
-
-TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrderedFromBusyCache) {
- BatchedCommandRequest request([&] {
- write_ops::InsertCommandRequest insertOp(nss);
- insertOp.setWriteCommandRequestBase([] {
- write_ops::WriteCommandRequestBase writeCommandBase;
- writeCommandBase.setOrdered(true);
- return writeCommandBase;
- }());
- insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
- return insertOp;
- }());
-
- auto future = launchAsync([&] {
- BatchedCommandResponse response;
- BatchWriteExecStats stats;
- BatchWriteExec::executeBatch(
- operationContext(), singleShardNSTargeter, request, &response, &stats);
-
- ASSERT(response.isErrDetailsSet());
- ASSERT_GT(response.sizeErrDetails(), 0u);
- ASSERT_EQ(ErrorCodes::ShardCannotRefreshDueToLocksHeld,
- response.getErrDetailsAt(0).getStatus().code());
- });
-
- expectInsertsReturnCannotRefreshErrors({BSON("x" << 1), BSON("x" << 2)});
-
- future.default_timed_get();
-}
-
TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientTxnError) {
BatchedCommandRequest request([&] {
write_ops::InsertCommandRequest insertOp(nss);
@@ -2837,13 +2408,16 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientTxnError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
BatchWriteExecStats stats;
- ASSERT_THROWS(BatchWriteExec::executeBatch(
- operationContext(), singleShardNSTargeter, request, &response, &stats),
- WriteConflictException);
+ ASSERT_THROWS_CODE(
+ BatchWriteExec::executeBatch(
+ operationContext(), singleShardNSTargeter, request, &response, &stats),
+ AssertionException,
+ ErrorCodes::WriteConflict);
});
expectInsertsReturnTransientTxnErrors({BSON("x" << 1), BSON("x" << 2)});
@@ -2862,6 +2436,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_DispatchError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2893,6 +2468,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientDispatchError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
diff --git a/src/mongo/s/write_ops/batch_write_op.cpp b/src/mongo/s/write_ops/batch_write_op.cpp
index ed67f6c0088..a61ee3dd4bf 100644
--- a/src/mongo/s/write_ops/batch_write_op.cpp
+++ b/src/mongo/s/write_ops/batch_write_op.cpp
@@ -59,6 +59,7 @@ struct WriteErrorComp {
// batches before serializing.
//
// TODO: Revisit when we revisit command limits in general
+const int kEstUpdateOverheadBytes = (BSONObjMaxInternalSize - BSONObjMaxUserSize) / 100;
const int kEstDeleteOverheadBytes = (BSONObjMaxInternalSize - BSONObjMaxUserSize) / 100;
/**
@@ -158,17 +159,51 @@ int getWriteSizeBytes(const WriteOp& writeOp) {
return item.getDocument().objsize();
} else if (batchType == BatchedCommandRequest::BatchType_Update) {
// Note: Be conservative here - it's okay if we send slightly too many batches.
- const auto& update = item.getUpdate();
- auto estSize = write_ops::getUpdateSizeEstimate(update.getQ(),
- update.getU(),
- update.getC(),
- update.getUpsertSupplied().has_value(),
- update.getCollation(),
- update.getArrayFilters(),
- update.getHint());
+ auto estSize = static_cast<int>(BSONObj::kMinBSONLength);
+ static const auto boolSize = 1;
+
+ // Add the size of the 'collation' field, if present.
+ estSize += !item.getUpdate().getCollation() ? 0
+ : (UpdateOpEntry::kCollationFieldName.size() +
+ item.getUpdate().getCollation()->objsize());
+
+ // Add the size of the 'arrayFilters' field, if present.
+ estSize += !item.getUpdate().getArrayFilters() ? 0 : ([&item]() {
+ auto size = BSONObj::kMinBSONLength + UpdateOpEntry::kArrayFiltersFieldName.size();
+ for (auto&& filter : *item.getUpdate().getArrayFilters()) {
+ size += filter.objsize();
+ }
+ return size;
+ })();
+
+ // Add the sizes of the 'multi' and 'upsert' fields.
+ estSize += UpdateOpEntry::kUpsertFieldName.size() + boolSize;
+ estSize += UpdateOpEntry::kMultiFieldName.size() + boolSize;
+
+ // Add the size of 'upsertSupplied' field if present.
+ if (auto upsertSupplied = item.getUpdate().getUpsertSupplied()) {
+ estSize += UpdateOpEntry::kUpsertSuppliedFieldName.size() + boolSize;
+ }
+
+ // Add the sizes of the 'q' and 'u' fields.
+ estSize += (UpdateOpEntry::kQFieldName.size() + item.getUpdate().getQ().objsize() +
+ UpdateOpEntry::kUFieldName.size() + item.getUpdate().getU().objsize());
+
+ // Add the size of the 'c' field if present.
+ if (auto constants = item.getUpdate().getC()) {
+ estSize += UpdateOpEntry::kCFieldName.size() + item.getUpdate().getC()->objsize();
+ }
+
+ // Add the size of 'hint' field if present.
+ if (auto hint = item.getUpdate().getHint(); !hint.isEmpty()) {
+ estSize += UpdateOpEntry::kHintFieldName.size() + hint.objsize();
+ }
+
+ // Finally, add the constant updateOp overhead size.
+ estSize += kEstUpdateOverheadBytes;
// When running a debug build, verify that estSize is at least the BSON serialization size.
- dassert(estSize >= update.toBSON().objsize());
+ dassert(estSize >= item.getUpdate().toBSON().objsize());
return estSize;
} else if (batchType == BatchedCommandRequest::BatchType_Delete) {
// Note: Be conservative here - it's okay if we send slightly too many batches.
@@ -568,9 +603,6 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch&
wcb.setStmtIds(std::move(stmtIdsForOp));
}
- wcb.setBypassEmptyTsReplacement(
- _clientRequest.getWriteCommandRequestBase().getBypassEmptyTsReplacement());
-
return wcb;
}());
@@ -583,6 +615,19 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch&
if (dbVersion)
request.setDbVersion(*dbVersion);
+ if (_clientRequest.hasWriteConcern()) {
+ if (_clientRequest.isVerboseWC()) {
+ request.setWriteConcern(_clientRequest.getWriteConcern());
+ } else {
+ // Mongos needs to send to the shard with w > 0 so it will be able to see the
+ // writeErrors
+ request.setWriteConcern(upgradeWriteConcern(_clientRequest.getWriteConcern()));
+ }
+ } else if (!TransactionRouter::get(_opCtx)) {
+ // Apply the WC from the opCtx (except if in a transaction).
+ request.setWriteConcern(_opCtx->getWriteConcern().toBSON());
+ }
+
return request;
}
@@ -765,7 +810,7 @@ void BatchWriteOp::buildClientResponse(BatchedCommandResponse* batchResp) {
batchResp->setStatus(Status::OK());
// For non-verbose, it's all we need.
- if (!_opCtx->getWriteConcern().requiresWriteAcknowledgement()) {
+ if (!_clientRequest.isVerboseWC()) {
return;
}
@@ -816,7 +861,12 @@ void BatchWriteOp::buildClientResponse(BatchedCommandResponse* batchResp) {
}
}
- if (!_wcErrors.empty()) {
+ // Only return a write concern error if everything succeeded (unordered or ordered)
+ // OR if something succeeded and we're unordered
+ const bool orderedOps = _clientRequest.getWriteCommandRequestBase().getOrdered();
+ const bool reportWCError =
+ errOps.empty() || (!orderedOps && errOps.size() < _clientRequest.sizeWriteOps());
+ if (!_wcErrors.empty() && reportWCError) {
WriteConcernErrorDetail* error = new WriteConcernErrorDetail;
// Generate the multi-error message below
diff --git a/src/mongo/s/write_ops/batch_write_op_test.cpp b/src/mongo/s/write_ops/batch_write_op_test.cpp
index b21e64c6d53..bfda09f0814 100644
--- a/src/mongo/s/write_ops/batch_write_op_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_op_test.cpp
@@ -32,7 +32,6 @@
#include "mongo/s/concurrency/locker_mongos_client_observer.h"
#include "mongo/s/mock_ns_targeter.h"
#include "mongo/s/session_catalog_router.h"
-#include "mongo/s/shard_cannot_refresh_due_to_locks_held_exception.h"
#include "mongo/s/sharding_router_test_fixture.h"
#include "mongo/s/transaction_router.h"
#include "mongo/s/write_ops/batch_write_op.h"
@@ -235,6 +234,7 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSON("w" << 3));
BatchWriteOp batchOp(_opCtx, request);
@@ -246,6 +246,7 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) {
BatchedCommandRequest targetBatch =
batchOp.buildBatchRequest(*targeted.begin()->second, targeter);
+ ASSERT(targetBatch.getWriteConcern().woCompare(request.getWriteConcern()) == 0);
BatchedCommandResponse response;
buildResponse(1, &response);
@@ -304,22 +305,12 @@ TEST_F(BatchWriteOpTest, SingleStaleError) {
batchOp.noteBatchResponse(*targeted.begin()->second, response, nullptr);
ASSERT(!batchOp.isFinished());
- // Respond with a ShardCannotRefreshDueToLocksHeld error; the batch should still be retriable.
- targeted.clear();
- ASSERT_OK(batchOp.targetBatch(targeter, false, &targeted));
- buildResponse(0, &response);
- response.addToErrDetails(write_ops::WriteError(
- 0, Status{ShardCannotRefreshDueToLocksHeldInfo(nss), "mock cache busy error"}));
-
- batchOp.noteBatchResponse(*targeted.begin()->second, response, nullptr);
- ASSERT(!batchOp.isFinished());
-
- // Respond with an 'ok' response
targeted.clear();
ASSERT_OK(batchOp.targetBatch(targeter, false, &targeted));
buildResponse(1, &response);
+ // Respond with an 'ok' response
batchOp.noteBatchResponse(*targeted.begin()->second, response, nullptr);
ASSERT(batchOp.isFinished());
@@ -1010,7 +1001,9 @@ TEST_F(BatchWriteOpTest, MultiOpPartialSingleShardErrorOrdered) {
// Tests of edge-case functionality, lifecycle is assumed to be behaving normally
//
-// Multi-op (unordered) error and write concern error test.
+// Multi-op (unordered) error and write concern error test. We never report the write concern error
+// for single-doc batches, since the error means there's no write concern applied. Don't suppress
+// the error if ordered : false.
TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) {
NamespaceString nss("foo.bar");
ShardEndpoint endpoint(ShardId("shard"), ChunkVersion::IGNORED(), boost::none);
@@ -1027,6 +1020,7 @@ TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 1)});
return insertOp;
}());
+ request.setWriteConcern(BSON("w" << 3));
BatchWriteOp batchOp(_opCtx, request);
@@ -1051,7 +1045,8 @@ TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) {
ASSERT(clientResponse.isWriteConcernErrorSet());
}
-// Single-op (ordered) error and write concern error test.
+// Single-op (ordered) error and write concern error test. Suppress the write concern error if
+// ordered and we also have an error
TEST_F(BatchWriteOpTest, SingleOpErrorAndWriteConcernErrorOrdered) {
NamespaceString nss("foo.bar");
ShardEndpoint endpointA(ShardId("shardA"), ChunkVersion::IGNORED(), boost::none);
@@ -1069,6 +1064,7 @@ TEST_F(BatchWriteOpTest, SingleOpErrorAndWriteConcernErrorOrdered) {
updateOp.setUpdates({buildUpdate(BSON("x" << GTE << -1 << LT << 2), true)});
return updateOp;
}());
+ request.setWriteConcern(BSON("w" << 3));
BatchWriteOp batchOp(_opCtx, request);
@@ -1095,14 +1091,14 @@ TEST_F(BatchWriteOpTest, SingleOpErrorAndWriteConcernErrorOrdered) {
ASSERT(batchOp.isFinished());
ASSERT(++targetedIt == targeted.end());
- // Ordered reports write concern error.
+ // Ordered doesn't report write concern error
BatchedCommandResponse clientResponse;
batchOp.buildClientResponse(&clientResponse);
ASSERT(clientResponse.getOk());
ASSERT_EQUALS(clientResponse.getN(), 1);
ASSERT(clientResponse.isErrDetailsSet());
ASSERT_EQUALS(clientResponse.sizeErrDetails(), 1u);
- ASSERT(clientResponse.isWriteConcernErrorSet());
+ ASSERT(!clientResponse.isWriteConcernErrorSet());
}
// Targeting failure on second op in batch op (ordered)
@@ -1411,6 +1407,7 @@ TEST_F(BatchWriteOpTest, MultiOpTwoWCErrors) {
insertOp.setDocuments({BSON("x" << -1), BSON("x" << 2)});
return insertOp;
}());
+ request.setWriteConcern(BSON("w" << 3));
BatchWriteOp batchOp(_opCtx, request);
diff --git a/src/mongo/s/write_ops/batched_command_request.cpp b/src/mongo/s/write_ops/batched_command_request.cpp
index 185857d6acc..107f1a49204 100644
--- a/src/mongo/s/write_ops/batched_command_request.cpp
+++ b/src/mongo/s/write_ops/batched_command_request.cpp
@@ -38,6 +38,8 @@
namespace mongo {
namespace {
+const auto kWriteConcern = "writeConcern"_sd;
+
template <class T>
BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request) {
auto batchRequest = BatchedCommandRequest{T::parse(request)};
@@ -51,6 +53,11 @@ BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request
batchRequest.setShardVersion(shardVersion);
}
+ auto writeConcernField = request.body[kWriteConcern];
+ if (!writeConcernField.eoo()) {
+ batchRequest.setWriteConcern(writeConcernField.Obj());
+ }
+
// The 'isTimeseriesNamespace' is an internal parameter used for communication between mongos
// and mongod.
auto isTimeseriesNamespace =
@@ -166,9 +173,19 @@ const boost::optional<BSONObj>& BatchedCommandRequest::getLet() const {
return _visit(Visitor{});
};
-const OptionalBool& BatchedCommandRequest::getBypassEmptyTsReplacement() const {
- return _visit([](auto&& op) -> decltype(auto) { return op.getBypassEmptyTsReplacement(); });
-};
+bool BatchedCommandRequest::isVerboseWC() const {
+ if (!hasWriteConcern()) {
+ return true;
+ }
+
+ BSONObj writeConcern = getWriteConcern();
+ BSONElement wElem = writeConcern["w"];
+ if (!wElem.isNumber() || wElem.Number() != 0) {
+ return true;
+ }
+
+ return false;
+}
const write_ops::WriteCommandRequestBase& BatchedCommandRequest::getWriteCommandRequestBase()
const {
@@ -189,6 +206,10 @@ void BatchedCommandRequest::serialize(BSONObjBuilder* builder) const {
if (_dbVersion) {
builder->append("databaseVersion", _dbVersion->toBSON());
}
+
+ if (_writeConcern) {
+ builder->append(kWriteConcern, *_writeConcern);
+ }
}
BSONObj BatchedCommandRequest::toBSON() const {
diff --git a/src/mongo/s/write_ops/batched_command_request.h b/src/mongo/s/write_ops/batched_command_request.h
index 0bcb51a3556..eea7f7bbe11 100644
--- a/src/mongo/s/write_ops/batched_command_request.h
+++ b/src/mongo/s/write_ops/batched_command_request.h
@@ -52,25 +52,15 @@ public:
: _batchType(BatchType_Insert),
_insertReq(std::make_unique<write_ops::InsertCommandRequest>(std::move(insertOp))) {}
- BatchedCommandRequest(std::unique_ptr<write_ops::InsertCommandRequest> insertOp)
- : _batchType(BatchType_Insert), _insertReq(std::move(insertOp)) {}
-
BatchedCommandRequest(write_ops::UpdateCommandRequest updateOp)
: _batchType(BatchType_Update),
_updateReq(std::make_unique<write_ops::UpdateCommandRequest>(std::move(updateOp))) {}
- BatchedCommandRequest(std::unique_ptr<write_ops::UpdateCommandRequest> updateOp)
- : _batchType(BatchType_Update), _updateReq(std::move(updateOp)) {}
-
BatchedCommandRequest(write_ops::DeleteCommandRequest deleteOp)
: _batchType(BatchType_Delete),
_deleteReq(std::make_unique<write_ops::DeleteCommandRequest>(std::move(deleteOp))) {}
- BatchedCommandRequest(std::unique_ptr<write_ops::DeleteCommandRequest> deleteOp)
- : _batchType(BatchType_Delete), _deleteReq(std::move(deleteOp)) {}
-
BatchedCommandRequest(BatchedCommandRequest&&) = default;
- BatchedCommandRequest& operator=(BatchedCommandRequest&&) = default;
static BatchedCommandRequest parseInsert(const OpMsgRequest& request);
static BatchedCommandRequest parseUpdate(const OpMsgRequest& request);
@@ -101,19 +91,26 @@ public:
return *_deleteReq;
}
- std::unique_ptr<write_ops::InsertCommandRequest> extractInsertRequest() {
- return std::move(_insertReq);
+ std::size_t sizeWriteOps() const;
+
+ void setWriteConcern(const BSONObj& writeConcern) {
+ _writeConcern = writeConcern.getOwned();
}
- std::unique_ptr<write_ops::UpdateCommandRequest> extractUpdateRequest() {
- return std::move(_updateReq);
+ void unsetWriteConcern() {
+ _writeConcern = boost::none;
}
- std::unique_ptr<write_ops::DeleteCommandRequest> extractDeleteRequest() {
- return std::move(_deleteReq);
+ bool hasWriteConcern() const {
+ return _writeConcern.is_initialized();
}
- std::size_t sizeWriteOps() const;
+ const BSONObj& getWriteConcern() const {
+ invariant(_writeConcern);
+ return *_writeConcern;
+ }
+
+ bool isVerboseWC() const;
void setShardVersion(ChunkVersion shardVersion) {
_shardVersion = std::move(shardVersion);
@@ -149,7 +146,6 @@ public:
const boost::optional<LegacyRuntimeConstants>& getLegacyRuntimeConstants() const;
const boost::optional<BSONObj>& getLet() const;
- const OptionalBool& getBypassEmptyTsReplacement() const;
const write_ops::WriteCommandRequestBase& getWriteCommandRequestBase() const;
void setWriteCommandRequestBase(write_ops::WriteCommandRequestBase writeCommandBase);
@@ -232,6 +228,8 @@ private:
boost::optional<ChunkVersion> _shardVersion;
boost::optional<DatabaseVersion> _dbVersion;
+
+ boost::optional<BSONObj> _writeConcern;
};
/**
diff --git a/src/mongo/s/write_ops/batched_command_request_test.cpp b/src/mongo/s/write_ops/batched_command_request_test.cpp
index be0728b1533..9a5e968f10d 100644
--- a/src/mongo/s/write_ops/batched_command_request_test.cpp
+++ b/src/mongo/s/write_ops/batched_command_request_test.cpp
@@ -92,12 +92,14 @@ TEST(BatchedCommandRequest, InsertCloneWithIds) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
+ batchedRequest.setWriteConcern(BSON("w" << 2));
const auto clonedRequest(BatchedCommandRequest::cloneInsertWithIds(std::move(batchedRequest)));
ASSERT_EQ("xyz.abc", clonedRequest.getNS().ns());
ASSERT(clonedRequest.getWriteCommandRequestBase().getOrdered());
ASSERT(clonedRequest.getWriteCommandRequestBase().getBypassDocumentValidation());
+ ASSERT_BSONOBJ_EQ(BSON("w" << 2), clonedRequest.getWriteConcern());
const auto& insertDocs = clonedRequest.getInsertRequest().getDocuments();
ASSERT_EQ(2u, insertDocs.size());
diff --git a/src/mongo/s/write_ops/write_op.cpp b/src/mongo/s/write_ops/write_op.cpp
index 6b33d3b9312..236c7efbc94 100644
--- a/src/mongo/s/write_ops/write_op.cpp
+++ b/src/mongo/s/write_ops/write_op.cpp
@@ -29,24 +29,7 @@
#include "mongo/s/write_ops/write_op.h"
-
-#include <absl/container/flat_hash_set.h>
-#include <algorithm>
-#include <boost/move/utility_core.hpp>
-#include <boost/none.hpp>
-#include <boost/optional/optional.hpp>
-#include <ostream>
-#include <string>
-
-#include "mongo/base/error_codes.h"
-#include "mongo/base/status.h"
-#include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/db/catalog/collection_uuid_mismatch_info.h"
-#include "mongo/db/stats/counters.h"
-#include "mongo/s/sharding_feature_flags_gen.h"
#include "mongo/s/transaction_router.h"
-#include "mongo/s/write_ops/batch_write_op.h"
-#include "mongo/s/write_ops/batched_command_request.h"
#include "mongo/util/assert_util.h"
namespace mongo {
@@ -96,16 +79,6 @@ write_ops::WriteError combineOpErrors(const std::vector<ChildWriteOp const*>& er
Status(MultipleErrorsOccurredInfo(errB.arr()), msg.str()));
}
-bool isSafeToIgnoreErrorInPartiallyAppliedOp(write_ops::WriteError& error) {
- // UUID mismatch errors are safe to ignore if the actualCollection is null in conjuntion with
- // other successful operations. This is true because it means we wrongly targeted a non-owning
- // shard with the operation and we wouldn't have applied any modifications anyway.
- //
- // Note this is only safe if we're using ShardVersion::IGNORED since we're ignoring any
- // placement concern and broadcasting to all shards.
- return error.getStatus().code() == ErrorCodes::CollectionUUIDMismatch &&
- !error.getStatus().extraInfo<CollectionUUIDMismatchInfo>()->actualCollection();
-}
} // namespace
const BatchItemRef& WriteOp::getWriteItem() const {
@@ -209,29 +182,7 @@ void WriteOp::_updateOpState() {
_state = WriteOpState_Ready;
} else if (!childErrors.empty()) {
_error = combineOpErrors(childErrors);
- bool isTargetingAllShardsWithSVIgnored =
- childErrors.front()
- ->endpoint->shardVersion
- .map([&](const auto& cv) { return ChunkVersion::isIgnoredVersion(cv); })
- .get_value_or(false);
- // There are errors that are safe to ignore if they were correctly applied to other shards
- // and we're using ShardVersion::IGNORED. They are safe to ignore as they can be interpreted
- // as no-ops if the shard response had been instead a successful result since they wouldn't
- // have modified any data. As a result, we can swallow the errors and treat them as a
- // successful operation.
- if (isTargetingAllShardsWithSVIgnored && isSafeToIgnoreErrorInPartiallyAppliedOp(*_error) &&
- !_successfulShardSet.empty()) {
- if (!hasPendingChild) {
- _error.reset();
- _state = WriteOpState_Completed;
- } else {
- // As this error is acceptable we wait until all other operations finish to take a
- // decision.
- return;
- }
- } else {
- _state = WriteOpState_Error;
- }
+ _state = WriteOpState_Error;
} else if (hasPendingChild && _inTxn) {
// Return early here since this means that there were no errors while in txn
// but there are still ops that have not yet finished.