Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

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()) {