summaryrefslogtreecommitdiff
path: root/src/mongo/db/ops/write_ops_exec.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/ops/write_ops_exec.cpp')
-rw-r--r--src/mongo/db/ops/write_ops_exec.cpp159
1 files changed, 39 insertions, 120 deletions
diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp
index e58146ac5cd..f7fc2a84efd 100644
--- a/src/mongo/db/ops/write_ops_exec.cpp
+++ b/src/mongo/db/ops/write_ops_exec.cpp
@@ -44,7 +44,7 @@
#include "mongo/db/catalog/document_validation.h"
#include "mongo/db/catalog_raii.h"
#include "mongo/db/commands.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
#include "mongo/db/curop_failpoint_helpers.h"
#include "mongo/db/curop_metrics.h"
#include "mongo/db/dbhelpers.h"
@@ -119,51 +119,6 @@ MONGO_FAIL_POINT_DEFINE(hangWithLockDuringBatchUpdate);
MONGO_FAIL_POINT_DEFINE(hangWithLockDuringBatchRemove);
MONGO_FAIL_POINT_DEFINE(failAtomicTimeseriesWrites);
-
-/**
- * Metrics group for the `updateMany` and `deleteMany` operations. For each
- * operation, the `duration` and `numDocs` will contribute to aggregated total
- * and max metrics.
- */
-class MultiUpdateDeleteMetrics {
-public:
- void operator()(Microseconds duration, size_t numDocs) {
- _durationTotalMicroseconds.increment(durationCount<Microseconds>(duration));
- _durationTotalMs.set(
- durationCount<Milliseconds>(Microseconds{_durationTotalMicroseconds.get()}));
- _durationMaxMs.setIfMax(durationCount<Milliseconds>(duration));
-
- _numDocsTotal.increment(numDocs);
- _numDocsMax.setIfMax(numDocs);
- }
-
-private:
- /**
- * To avoid rapid accumulation of roundoff error in the duration total, it
- * is maintained precisely, and we arrange for the corresponding
- * Millisecond metric to hold an exported low-res image of it.
- */
- Counter64 _durationTotalMicroseconds;
-
- Atomic64Metric _durationTotalMs;
- ServerStatusMetricField<Atomic64Metric> _displayDurationTotalMs{
- "query.updateDeleteManyDurationTotalMs", &_durationTotalMs};
- Atomic64Metric _durationMaxMs;
- ServerStatusMetricField<Atomic64Metric> _displayDurationMaxMs{
- "query.updateDeleteManyDurationMaxMs", &_durationMaxMs};
-
- Counter64 _numDocsTotal;
- ServerStatusMetricField<Counter64> displayNumDocsTotal{
- "query.updateDeleteManyDocumentsTotalCount", &_numDocsTotal};
-
- Atomic64Metric _numDocsMax;
- ServerStatusMetricField<Atomic64Metric> _displayNumDocsMax{
- "query.updateDeleteManyDocumentsMaxCount", &_numDocsMax};
-};
-
-MultiUpdateDeleteMetrics collectMultiUpdateDeleteMetrics;
-
-
void updateRetryStats(OperationContext* opCtx, bool containsRetry) {
if (containsRetry) {
RetryableWritesStats::get(opCtx)->incrementRetriedCommandsCount();
@@ -174,7 +129,7 @@ void finishCurOp(OperationContext* opCtx, CurOp* curOp) {
try {
curOp->done();
auto executionTimeMicros = duration_cast<Microseconds>(curOp->elapsedTimeExcludingPauses());
- curOp->debug().additiveMetrics.executionTime = executionTimeMicros;
+ curOp->debug().executionTime = executionTimeMicros;
recordCurOpMetrics(opCtx);
Top::get(opCtx->getServiceContext())
@@ -338,20 +293,15 @@ bool handleError(OperationContext* opCtx,
return false;
}
- if (ex.code() == ErrorCodes::StaleDbVersion || ErrorCodes::isStaleShardVersionError(ex) ||
- ex.code() == ErrorCodes::ShardCannotRefreshDueToLocksHeld) {
+ if (ex.code() == ErrorCodes::StaleDbVersion || ErrorCodes::isStaleShardVersionError(ex)) {
if (!opCtx->getClient()->isInDirectClient()) {
auto& oss = OperationShardingState::get(opCtx);
oss.setShardingOperationFailedStatus(ex.toStatus());
}
- // For routing errors, it is guaranteed that all subsequent operations will fail
+ // Since this is a routing error, it is guaranteed that all subsequent operations will fail
// with the same cause, so don't try doing any more operations. The command reply serializer
// will handle repeating this error for unordered writes.
- // (On the other hand, ShardCannotRefreshDueToLocksHeld is caused by a temporary inability
- // to access a stable version of the cache during the execution of the batch; the error is
- // returned back to the router to leverage its capability of selectively retrying
- // operations).
out->results.emplace_back(ex.toStatus());
return false;
}
@@ -381,6 +331,10 @@ bool handleError(OperationContext* opCtx,
return false;
}
+ if (ex.code() == ErrorCodes::ShardCannotRefreshDueToLocksHeld) {
+ throw;
+ }
+
out->results.emplace_back(ex.toStatus());
return !wholeOp.getOrdered();
}
@@ -496,13 +450,8 @@ bool insertBatchAndHandleErrors(OperationContext* opCtx,
opCtx,
wholeOp.getNamespace(),
fixLockModeForSystemDotViewsChanges(wholeOp.getNamespace(), MODE_IX));
- checkCollectionUUIDMismatch(opCtx,
- wholeOp.getNamespace(),
- collection->getCollection(),
- wholeOp.getCollectionUUID());
- if (*collection) {
+ if (*collection)
break;
- }
if (source == OperationSource::kTimeseriesInsert) {
assertTimeseriesBucketsCollectionNotFound(wholeOp.getNamespace());
@@ -548,6 +497,11 @@ bool insertBatchAndHandleErrors(OperationContext* opCtx,
if (shouldProceedWithBatchInsert) {
try {
if (!collection->getCollection()->isCapped() && !inTxn && batch.size() > 1) {
+ checkCollectionUUIDMismatch(opCtx,
+ wholeOp.getNamespace(),
+ collection->getCollection(),
+ wholeOp.getCollectionUUID());
+
// First try doing it all together. If all goes well, this is all we need to do.
// See Collection::_insertDocuments for why we do all capped inserts one-at-a-time.
lastOpFixer->startingOp();
@@ -590,6 +544,10 @@ bool insertBatchAndHandleErrors(OperationContext* opCtx,
// Transactions are not allowed to operate on capped collections.
uassertStatusOK(
checkIfTransactionOnCappedColl(opCtx, collection->getCollection()));
+ checkCollectionUUIDMismatch(opCtx,
+ wholeOp.getNamespace(),
+ collection->getCollection(),
+ wholeOp.getCollectionUUID());
lastOpFixer->startingOp();
insertDocuments(opCtx,
collection->getCollection(),
@@ -673,6 +631,7 @@ bool getFleCrudProcessed(OperationContext* opCtx,
WriteResult performInserts(OperationContext* opCtx,
const write_ops::InsertCommandRequest& wholeOp,
OperationSource source) {
+
// Insert performs its own retries, so we should only be within a WriteUnitOfWork when run in a
// transaction.
auto txnParticipant = TransactionParticipant::get(opCtx);
@@ -731,20 +690,11 @@ WriteResult performInserts(OperationContext* opCtx,
const size_t maxBatchBytes = write_ops::insertVectorMaxBytes;
batch.reserve(std::min(wholeOp.getDocuments().size(), maxBatchSize));
- // If 'wholeOp.getBypassEmptyTsReplacement()' is true or if 'source' is 'kFromMigrate', set
- // "bypassEmptyTsReplacement=true" for fixDocumentForInsert().
- const bool bypassEmptyTsReplacement = (source == OperationSource::kFromMigrate) ||
- static_cast<bool>(wholeOp.getBypassEmptyTsReplacement());
-
for (auto&& doc : wholeOp.getDocuments()) {
const bool isLastDoc = (&doc == &wholeOp.getDocuments().back());
bool containsDotsAndDollarsField = false;
-
- auto fixedDoc = fixDocumentForInsert(
- opCtx, doc, bypassEmptyTsReplacement, &containsDotsAndDollarsField);
-
+ auto fixedDoc = fixDocumentForInsert(opCtx, doc, &containsDotsAndDollarsField);
const StmtId stmtId = getStmtIdForWriteOp(opCtx, wholeOp, stmtIdIndex++);
-
const bool wasAlreadyExecuted = opCtx->isRetryableWrite() &&
txnParticipant.checkStatementExecutedNoOplogEntryFetch(opCtx, stmtId);
@@ -846,7 +796,6 @@ static SingleWriteResult performSingleUpdateOp(OperationContext* opCtx,
boost::optional<AutoGetCollection> collection;
while (true) {
collection.emplace(opCtx, ns, fixLockModeForSystemDotViewsChanges(ns, MODE_IX));
- checkCollectionUUIDMismatch(opCtx, ns, collection->getCollection(), opCollectionUUID);
if (*collection) {
break;
}
@@ -911,6 +860,8 @@ static SingleWriteResult performSingleUpdateOp(OperationContext* opCtx,
uassertStatusOK(checkIfTransactionOnCappedColl(opCtx, coll));
}
+ checkCollectionUUIDMismatch(opCtx, ns, collection->getCollection(), opCollectionUUID);
+
const ExtensionsCallbackReal extensionsCallback(opCtx, &updateRequest->getNamespaceString());
ParsedUpdate parsedUpdate(opCtx, updateRequest, extensionsCallback, forgoOpCounterIncrements);
uassertStatusOK(parsedUpdate.parseRequest());
@@ -986,7 +937,6 @@ static SingleWriteResult performSingleUpdateOpWithDupKeyRetry(
const write_ops::UpdateOpEntry& op,
LegacyRuntimeConstants runtimeConstants,
const boost::optional<BSONObj>& letParams,
- const OptionalBool& bypassEmptyTsReplacement,
OperationSource source,
bool forgoOpCounterIncrements) {
globalOpCounters.gotUpdate();
@@ -1015,9 +965,10 @@ static SingleWriteResult performSingleUpdateOpWithDupKeyRetry(
if (letParams) {
request.setLetParameters(std::move(letParams));
}
- request.setBypassEmptyTsReplacement(bypassEmptyTsReplacement);
request.setStmtIds(stmtIds);
- request.setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+ request.setYieldPolicy(opCtx->inMultiDocumentTransaction()
+ ? PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY
+ : PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
request.setSource(source);
size_t numAttempts = 0;
@@ -1144,12 +1095,7 @@ WriteResult performUpdates(OperationContext* opCtx,
? *wholeOp.getStmtIds()
: std::vector<StmtId>{stmtId};
- boost::optional<Timer> timer;
- if (singleOp.getMulti()) {
- timer.emplace();
- }
-
- const SingleWriteResult&& reply =
+ out.results.emplace_back(
performSingleUpdateOpWithDupKeyRetry(opCtx,
ns,
wholeOp.getCollectionUUID(),
@@ -1157,27 +1103,11 @@ WriteResult performUpdates(OperationContext* opCtx,
singleOp,
runtimeConstants,
wholeOp.getLet(),
- wholeOp.getBypassEmptyTsReplacement(),
source,
- forgoOpCounterIncrements);
- out.results.emplace_back(reply);
+ forgoOpCounterIncrements));
forgoOpCounterIncrements = true;
lastOpFixer.finishedOpSuccessfully();
-
- if (singleOp.getMulti()) {
- updateManyCount.increment(1);
- collectMultiUpdateDeleteMetrics(timer->elapsed(), reply.getNModified());
- }
} catch (const DBException& ex) {
- // Do not handle errors for time-series bucket compressions. They need to be transparent
- // to users to not interfere with any decisions around operation retry. It is OK to
- // leave bucket uncompressed in these edge cases. We just record the status to the
- // result vector so we can keep track of statistics for failed bucket compressions.
- if (source == OperationSource::kTimeseriesBucketCompression) {
- out.results.emplace_back(ex.toStatus());
- break;
- }
-
out.canContinue = handleError(
opCtx, ex, ns, wholeOp.getWriteCommandRequestBase(), singleOp.getMulti(), &out);
if (!out.canContinue) {
@@ -1223,7 +1153,9 @@ static SingleWriteResult performSingleDeleteOp(OperationContext* opCtx,
request.setQuery(op.getQ());
request.setCollation(write_ops::collationOf(op));
request.setMulti(op.getMulti());
- request.setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+ request.setYieldPolicy(opCtx->inMultiDocumentTransaction()
+ ? PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY
+ : PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
request.setStmtId(stmtId);
request.setHint(op.getHint());
@@ -1390,28 +1322,15 @@ WriteResult performDeletes(OperationContext* opCtx,
});
try {
lastOpFixer.startingOp();
-
- boost::optional<Timer> timer;
- if (singleOp.getMulti()) {
- timer.emplace();
- }
-
- const SingleWriteResult&& reply = performSingleDeleteOp(opCtx,
- ns,
- wholeOp.getCollectionUUID(),
- stmtId,
- singleOp,
- runtimeConstants,
- wholeOp.getLet(),
- source);
- out.results.push_back(reply);
+ out.results.push_back(performSingleDeleteOp(opCtx,
+ ns,
+ wholeOp.getCollectionUUID(),
+ stmtId,
+ singleOp,
+ runtimeConstants,
+ wholeOp.getLet(),
+ source));
lastOpFixer.finishedOpSuccessfully();
-
- // Collect metrics.
- if (singleOp.getMulti()) {
- deleteManyCount.increment(1);
- collectMultiUpdateDeleteMetrics(timer->elapsed(), reply.getN());
- }
} catch (const DBException& ex) {
out.canContinue = handleError(
opCtx, ex, ns, wholeOp.getWriteCommandRequestBase(), false /* multiUpdate */, &out);
@@ -1508,7 +1427,7 @@ Status performAtomicTimeseriesWrites(
doc_diff::applyDiff(original.value(),
update.getU().getDiff(),
&CollectionQueryInfo::get(*coll).getIndexKeys(opCtx),
- update.getU().mustCheckExistenceForInsertOperations());
+ static_cast<bool>(repl::tenantMigrationRecipientInfo(opCtx)));
CollectionUpdateArgs args;
if (const auto& stmtIds = op.getStmtIds()) {
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 11:36:55 +0000