summaryrefslogtreecommitdiff
path: root/src/mongo/s/write_ops
diff options
context:
space:
mode:
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.cpp285
-rw-r--r--src/mongo/s/write_ops/batch_write_op.cpp16
-rw-r--r--src/mongo/s/write_ops/batch_write_op_test.cpp5
-rw-r--r--src/mongo/s/write_ops/batched_command_request.cpp32
-rw-r--r--src/mongo/s/write_ops/batched_command_request.h22
-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, 324 insertions, 113 deletions
diff --git a/src/mongo/s/write_ops/batch_write_exec.cpp b/src/mongo/s/write_ops/batch_write_exec.cpp
index 45b89579dbb..980d376b607 100644
--- a/src/mongo/s/write_ops/batch_write_exec.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec.cpp
@@ -104,6 +104,25 @@ 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,
@@ -195,6 +214,7 @@ 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) {
@@ -217,6 +237,10 @@ 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 a2dc99859cc..3e90e5b65d3 100644
--- a/src/mongo/s/write_ops/batch_write_exec_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_exec_test.cpp
@@ -397,7 +397,6 @@ 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([&] {
@@ -436,7 +435,6 @@ TEST_F(BatchWriteExecTest, SingleUpdateTargetsShardWithLet) {
<< "100")))});
return updateOp;
}());
- updateRequest.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
const static Timestamp timestamp(2);
@@ -522,7 +520,6 @@ TEST_F(BatchWriteExecTest, SingleDeleteTargetsShardWithLet) {
deleteOp.setDeletes(std::vector{write_ops::DeleteOpEntry(q, false)});
return deleteOp;
}());
- deleteRequest.setWriteConcern(BSONObj());
const static auto epoch = OID::gen();
@@ -608,7 +605,6 @@ TEST_F(BatchWriteExecTest, MultiOpLargeOrdered) {
insertOp.setDocuments(docsToInsert);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -641,7 +637,6 @@ TEST_F(BatchWriteExecTest, SingleOpUnorderedError) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -682,7 +677,6 @@ TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithStaleShardVersionError) {
insertOp.setDocuments(docsToInsert);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -714,7 +708,6 @@ 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);
@@ -811,7 +804,6 @@ TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithCannotRefreshError) {
insertOp.setDocuments(docsToInsert);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -848,7 +840,6 @@ 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);
@@ -952,7 +943,6 @@ 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);
@@ -1066,7 +1056,6 @@ 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);
@@ -1176,7 +1165,6 @@ 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);
@@ -1276,7 +1264,6 @@ TEST_F(BatchWriteExecTest, StaleShardOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -1308,7 +1295,6 @@ TEST_F(BatchWriteExecTest, MultiStaleShardOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1346,7 +1332,6 @@ TEST_F(BatchWriteExecTest, TooManyStaleShardOp) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1381,7 +1366,6 @@ TEST_F(BatchWriteExecTest, StaleDbOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -1413,7 +1397,6 @@ TEST_F(BatchWriteExecTest, MultiStaleDbOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1451,7 +1434,6 @@ TEST_F(BatchWriteExecTest, TooManyStaleDbOp) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1486,7 +1468,6 @@ TEST_F(BatchWriteExecTest, MultiCannotRefreshShardOp) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1522,7 +1503,6 @@ TEST_F(BatchWriteExecTest, TooManyCannotRefreshShardOp) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -1566,7 +1546,6 @@ TEST_F(BatchWriteExecTest, RetryableWritesLargeBatch) {
insertOp.setDocuments(docsToInsert);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1601,7 +1580,6 @@ 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"});
@@ -1640,7 +1618,6 @@ TEST_F(BatchWriteExecTest, RetryableErrorTxnNumber) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1678,7 +1655,6 @@ TEST_F(BatchWriteExecTest, NonRetryableErrorTxnNumber) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1720,7 +1696,6 @@ TEST_F(BatchWriteExecTest, StaleEpochIsNotRetryable) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest());
operationContext()->setTxnNumber(5);
@@ -1748,6 +1723,249 @@ 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([&] {
@@ -1760,7 +1978,6 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorOrderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -1791,7 +2008,6 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorUnorderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -1822,7 +2038,6 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorUnorderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
const int numTenantMigrationAbortedErrors = 3;
@@ -1857,7 +2072,6 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorOrderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
const int numTenantMigrationAbortedErrors = 3;
@@ -1892,7 +2106,6 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationAbortedErrorOrderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -1925,7 +2138,6 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationErrorUnorderedOp) {
insertOp.setDocuments(expected);
return insertOp;
}());
- request.setWriteConcern(BSONObj());
// Execute request
auto future = launchAsync([&] {
@@ -2010,7 +2222,6 @@ 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);
@@ -2146,7 +2357,6 @@ 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);
@@ -2290,7 +2500,6 @@ 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);
@@ -2481,7 +2690,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchThrows_CommandError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2513,7 +2721,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2543,7 +2750,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrdered) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2573,7 +2779,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorFromBusyCache)
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2603,7 +2808,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrderedFromBusy
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2633,7 +2837,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientTxnError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2659,7 +2862,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_DispatchError) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)});
return insertOp;
}());
- request.setWriteConcern(BSONObj());
auto future = launchAsync([&] {
BatchedCommandResponse response;
@@ -2691,7 +2893,6 @@ 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 f93e5302e4f..ed67f6c0088 100644
--- a/src/mongo/s/write_ops/batch_write_op.cpp
+++ b/src/mongo/s/write_ops/batch_write_op.cpp
@@ -568,6 +568,9 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch&
wcb.setStmtIds(std::move(stmtIdsForOp));
}
+ wcb.setBypassEmptyTsReplacement(
+ _clientRequest.getWriteCommandRequestBase().getBypassEmptyTsReplacement());
+
return wcb;
}());
@@ -580,19 +583,6 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch&
if (dbVersion)
request.setDbVersion(*dbVersion);
- if (_clientRequest.hasWriteConcern()) {
- if (_clientRequest.requiresWriteAcknowledgement()) {
- 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;
}
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 80ea3818494..b21e64c6d53 100644
--- a/src/mongo/s/write_ops/batch_write_op_test.cpp
+++ b/src/mongo/s/write_ops/batch_write_op_test.cpp
@@ -235,7 +235,6 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) {
insertOp.setDocuments({BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSON("w" << 3));
BatchWriteOp batchOp(_opCtx, request);
@@ -247,7 +246,6 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) {
BatchedCommandRequest targetBatch =
batchOp.buildBatchRequest(*targeted.begin()->second, targeter);
- ASSERT(targetBatch.getWriteConcern().woCompare(request.getWriteConcern()) == 0);
BatchedCommandResponse response;
buildResponse(1, &response);
@@ -1029,7 +1027,6 @@ TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) {
insertOp.setDocuments({BSON("x" << 1), BSON("x" << 1)});
return insertOp;
}());
- request.setWriteConcern(BSON("w" << 3));
BatchWriteOp batchOp(_opCtx, request);
@@ -1072,7 +1069,6 @@ 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);
@@ -1415,7 +1411,6 @@ 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 e6b3c970592..185857d6acc 100644
--- a/src/mongo/s/write_ops/batched_command_request.cpp
+++ b/src/mongo/s/write_ops/batched_command_request.cpp
@@ -38,8 +38,6 @@
namespace mongo {
namespace {
-const auto kWriteConcern = "writeConcern"_sd;
-
template <class T>
BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request) {
auto batchRequest = BatchedCommandRequest{T::parse(request)};
@@ -53,16 +51,6 @@ BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request
batchRequest.setShardVersion(shardVersion);
}
- auto writeConcernField = request.body[kWriteConcern];
- if (!writeConcernField.eoo()) {
- auto wcObj = writeConcernField.Obj();
- // Client write concerns without 'w' fields should be filled with the default write concern,
- // which should be populated later to the operation context during the command setup phase.
- if (wcObj.hasElement("w")) {
- batchRequest.setWriteConcern(wcObj);
- }
- }
-
// The 'isTimeseriesNamespace' is an internal parameter used for communication between mongos
// and mongod.
auto isTimeseriesNamespace =
@@ -178,19 +166,9 @@ const boost::optional<BSONObj>& BatchedCommandRequest::getLet() const {
return _visit(Visitor{});
};
-bool BatchedCommandRequest::requiresWriteAcknowledgement() const {
- if (!hasWriteConcern()) {
- return true;
- }
-
- BSONObj writeConcern = getWriteConcern();
- BSONElement wElem = writeConcern["w"];
- if (!wElem.isNumber() || wElem.Number() != 0) {
- return true;
- }
-
- return false;
-}
+const OptionalBool& BatchedCommandRequest::getBypassEmptyTsReplacement() const {
+ return _visit([](auto&& op) -> decltype(auto) { return op.getBypassEmptyTsReplacement(); });
+};
const write_ops::WriteCommandRequestBase& BatchedCommandRequest::getWriteCommandRequestBase()
const {
@@ -211,10 +189,6 @@ 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 1834fe4286b..0bcb51a3556 100644
--- a/src/mongo/s/write_ops/batched_command_request.h
+++ b/src/mongo/s/write_ops/batched_command_request.h
@@ -115,25 +115,6 @@ public:
std::size_t sizeWriteOps() const;
- void setWriteConcern(const BSONObj& writeConcern) {
- _writeConcern = writeConcern.getOwned();
- }
-
- void unsetWriteConcern() {
- _writeConcern = boost::none;
- }
-
- bool hasWriteConcern() const {
- return _writeConcern.is_initialized();
- }
-
- const BSONObj& getWriteConcern() const {
- invariant(_writeConcern);
- return *_writeConcern;
- }
-
- bool requiresWriteAcknowledgement() const;
-
void setShardVersion(ChunkVersion shardVersion) {
_shardVersion = std::move(shardVersion);
}
@@ -168,6 +149,7 @@ 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);
@@ -250,8 +232,6 @@ 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 9a5e968f10d..be0728b1533 100644
--- a/src/mongo/s/write_ops/batched_command_request_test.cpp
+++ b/src/mongo/s/write_ops/batched_command_request_test.cpp
@@ -92,14 +92,12 @@ 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 236c7efbc94..6b33d3b9312 100644
--- a/src/mongo/s/write_ops/write_op.cpp
+++ b/src/mongo/s/write_ops/write_op.cpp
@@ -29,7 +29,24 @@
#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 {
@@ -79,6 +96,16 @@ 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 {
@@ -182,7 +209,29 @@ void WriteOp::_updateOpState() {
_state = WriteOpState_Ready;
} else if (!childErrors.empty()) {
_error = combineOpErrors(childErrors);
- _state = WriteOpState_Error;
+ 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;
+ }
} 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.
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 11:46:10 +0000