New upstream version 6.0.20upstream

21 files changed, 416 insertions, 208 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 9eae99e2838..d5210d9603c 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -232,8 +232,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/commands',
         '$BUILD_DIR/mongo/db/concurrency/exception_util',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/dbdirectclient',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/storage/backup_cursor_hooks',
         'fsync_locked',
     ]
@@ -387,7 +387,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/query/ce/query_ce',
         '$BUILD_DIR/mongo/db/query/command_request_response',
         '$BUILD_DIR/mongo/db/query/cursor_response_idl',
-        '$BUILD_DIR/mongo/db/query/optimizer/optimizer',
+        '$BUILD_DIR/mongo/db/query/query_shape/query_shape',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/repl/replica_set_messages',
         '$BUILD_DIR/mongo/db/repl/tenant_migration_access_blocker',
@@ -496,6 +497,7 @@ env.Library(
         'shutdown.idl',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/bson/bson_validate',
         '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/util/fail_point',
     ],
@@ -672,8 +674,8 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/matcher/expressions',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/server_options_core',
     ],
 )
diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp
index 26ee8ab5d4b..48d899efcd1 100644
--- a/src/mongo/db/commands/compact.cpp
+++ b/src/mongo/db/commands/compact.cpp
@@ -27,23 +27,34 @@
  *    it in the license file.
  */
 
+#include <absl/container/btree_set.h>
 #include <string>
 #include <vector>
 
 #include "mongo/db/auth/action_set.h"
 #include "mongo/db/auth/action_type.h"
+#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_compact.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/curop.h"
+#include "mongo/db/db_raii.h"
 #include "mongo/db/jsobj.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/repl/member_state.h"
 #include "mongo/db/repl/replication_coordinator.h"
 
 namespace mongo {
 
+namespace {
+static absl::btree_set<UUID> compactsRunning;
+}  // namespace
+
 using std::string;
 using std::stringstream;
 
@@ -65,6 +76,7 @@ public:
         actions.addAction(ActionType::compact);
         out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
     }
+
     std::string help() const override {
         return "compact collection\n"
                "warning: this operation locks the database and is slow. you can cancel with "
@@ -75,36 +87,59 @@ public:
     CompactCmd() : ErrmsgCommandDeprecated("compact") {}
 
     virtual bool errmsgRun(OperationContext* opCtx,
-                           const string& db,
+                           const std::string& dbName,
                            const BSONObj& cmdObj,
                            string& errmsg,
                            BSONObjBuilder& result) {
-        NamespaceString nss = CommandHelpers::parseNsCollectionRequired(db, cmdObj);
-
-        repl::ReplicationCoordinator* replCoord = repl::ReplicationCoordinator::get(opCtx);
-        if (replCoord->getMemberState().primary() && !cmdObj["force"].trueValue()) {
-            errmsg =
-                "will not run compact on an active replica set primary as this is a slow blocking "
-                "operation. use force:true to force";
-            return false;
-        }
+        NamespaceString collectionNss = CommandHelpers::parseNsCollectionRequired(dbName, cmdObj);
+
+        Lock::GlobalLock lk(opCtx,
+                            MODE_IX,
+                            Date_t::max(),
+                            Lock::InterruptBehavior::kThrow,
+                            /*skipRSTLLock=*/true);
+
+        // Hold reference to the catalog for collection lookup without locks to be safe.
+        auto collectionCatalog = CollectionCatalog::get(opCtx);
 
-        if (nss.isSystem()) {
-            // Items in system.* cannot be moved as there might be pointers to them.
-            errmsg = "can't compact a system namespace";
-            return false;
+        CollectionPtr collection = [&]() {
+            if (CollectionPtr collection = CollectionPtr(
+                    collectionCatalog->lookupCollectionByNamespace(opCtx, collectionNss))) {
+                return collection;
+            }
+
+            // Check if this is a time-series collection.
+            auto bucketsNs = collectionNss.makeTimeseriesBucketsNamespace();
+            if (CollectionPtr collection = CollectionPtr(
+                    collectionCatalog->lookupCollectionByNamespace(opCtx, bucketsNs))) {
+                return collection;
+            }
+
+            return CollectionPtr();
+        }();
+
+        if (!collection) {
+            std::shared_ptr<const ViewDefinition> view =
+                collectionCatalog->lookupView(opCtx, collectionNss);
+            uassert(ErrorCodes::CommandNotSupportedOnView, "can't compact a view", !view);
+            uasserted(ErrorCodes::NamespaceNotFound, "collection does not exist");
         }
 
-        // This command is internal to the storage engine and should not block oplog application.
-        ShouldNotConflictWithSecondaryBatchApplicationBlock noPBWMBlock(opCtx->lockState());
+        AutoStatsTracker statsTracker(opCtx,
+                                      collectionNss,
+                                      Top::LockType::NotLocked,
+                                      AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
+                                      collectionCatalog->getDatabaseProfileLevel(dbName));
+
+        StatusWith<int64_t> status = compactCollection(opCtx, collection);
 
-        StatusWith<int64_t> status = compactCollection(opCtx, nss);
         uassertStatusOK(status.getStatus());
 
         int64_t bytesFreed = status.getValue();
         if (bytesFreed < 0) {
-            // When compacting a collection that is actively being written to, it is possible that
-            // the collection is larger at the completion of compaction than when it started.
+            // When compacting a collection that is actively being written to, it is possible
+            // that the collection is larger at the completion of compaction than when it
+            // started.
             bytesFreed = 0;
         }
 
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index 7fb22de4c0f..57291c7b782 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -182,12 +182,8 @@ public:
 
             // An empty PrivilegeVector is acceptable because these privileges are only checked on
             // getMore and explain will not open a cursor.
-            return runAggregate(opCtx,
-                                viewAggRequest.getNamespace(),
-                                viewAggRequest,
-                                viewAggregation.getValue(),
-                                PrivilegeVector(),
-                                result);
+            return runAggregate(
+                opCtx, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result);
         }
 
         const auto& collection = ctx->getCollection();
@@ -236,6 +232,8 @@ public:
             &hangBeforeCollectionCount, opCtx, "hangBeforeCollectionCount", []() {}, nss);
 
         auto request = CountCommandRequest::parse(IDLParserErrorContext("count"), cmdObj);
+        auto curOp = CurOp::get(opCtx);
+        curOp->beginQueryPlanningTimer();
         if (shouldDoFLERewrite(request)) {
             processFLECountD(opCtx, nss, &request);
         }
@@ -285,7 +283,6 @@ public:
         auto exec = std::move(statusWithPlanExecutor.getValue());
 
         // Store the plan summary string in CurOp.
-        auto curOp = CurOp::get(opCtx);
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
             curOp->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
diff --git a/src/mongo/db/commands/current_op.cpp b/src/mongo/db/commands/current_op.cpp
index 78fec805202..b4a16bc40cb 100644
--- a/src/mongo/db/commands/current_op.cpp
+++ b/src/mongo/db/commands/current_op.cpp
@@ -77,12 +77,7 @@ public:
             privileges = {Privilege(ResourcePattern::forClusterResource(), ActionType::inprog)};
         }
 
-        auto status = runAggregate(opCtx,
-                                   request.getNamespace(),
-                                   request,
-                                   std::move(aggCmdObj),
-                                   privileges,
-                                   &replyBuilder);
+        auto status = runAggregate(opCtx, request, std::move(aggCmdObj), privileges, &replyBuilder);
 
         if (!status.isOK()) {
             return status;
diff --git a/src/mongo/db/commands/dbcheck.cpp b/src/mongo/db/commands/dbcheck.cpp
index ef4a566bc8f..5d4579691dd 100644
--- a/src/mongo/db/commands/dbcheck.cpp
+++ b/src/mongo/db/commands/dbcheck.cpp
@@ -53,6 +53,8 @@
 #include "mongo/logv2/log.h"
 
 MONGO_FAIL_POINT_DEFINE(sleepAfterExtraIndexKeysHashing);
+MONGO_FAIL_POINT_DEFINE(hangBeforeProcessingDbCheckRun);
+MONGO_FAIL_POINT_DEFINE(hangBeforeAddingDBCheckBatchToOplog);
 
 namespace mongo {
 
@@ -288,6 +290,11 @@ protected:
 
         DbCheckStartAndStopLogger startStop(opCtx);
 
+        if (MONGO_unlikely(hangBeforeProcessingDbCheckRun.shouldFail())) {
+            LOGV2(7949000, "Hanging dbcheck due to failpoint 'hangBeforeProcessingDbCheckRun'");
+            hangBeforeProcessingDbCheckRun.pauseWhileSet();
+        }
+
         for (const auto& coll : *_run) {
             try {
                 _doCollection(opCtx, coll);
@@ -440,11 +447,14 @@ private:
             WriteConcernResult unused;
             auto status = waitForWriteConcern(opCtx, stats.time, info.writeConcern, &unused);
             if (!status.isOK()) {
-                auto entry = dbCheckWarningHealthLogEntry(info.nss,
-                                                          "dbCheck failed waiting for writeConcern",
-                                                          OplogEntriesEnum::Batch,
-                                                          status);
+                // TODO SERVER-89817: Add context with batch ID and lastKey once those are
+                // backported.
+                auto entry = dbCheckErrorHealthLogEntry(info.nss,
+                                                        "dbCheck failed waiting for writeConcern",
+                                                        OplogEntriesEnum::Batch,
+                                                        status);
                 HealthLogInterface::get(opCtx)->log(*entry);
+                return;
             }
 
             start = stats.lastKey;
@@ -575,6 +585,12 @@ private:
                 batch.setMaxKey(BSONKey(hasher->lastKey()));
                 batch.setReadTimestamp(readTimestamp);
 
+                if (MONGO_unlikely(hangBeforeAddingDBCheckBatchToOplog.shouldFail())) {
+                    LOGV2(8589000,
+                          "Hanging dbCheck due to failpoint 'hangBeforeAddingDBCheckBatchToOplog'");
+                    hangBeforeAddingDBCheckBatchToOplog.pauseWhileSet();
+                }
+
                 // Send information on this batch over the oplog.
                 result.time = _logOp(opCtx, info.nss, collection->uuid(), batch.toBSON());
                 result.readTimestamp = readTimestamp;
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index f642ba2f49d..ef08e46aee7 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -172,7 +172,7 @@ public:
             // An empty PrivilegeVector is acceptable because these privileges are only checked on
             // getMore and explain will not open a cursor.
             return runAggregate(
-                opCtx, nss, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result);
+                opCtx, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result);
         }
 
         const auto& collection = ctx->getCollection();
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 8e0a48f013b..a2c773af89d 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -236,8 +236,8 @@ public:
             collection.getWritableCollection()->getIndexCatalog()->dropAllIndexes(
                 opCtx, collection.getWritableCollection(), true, {});
 
-            swIndexesToRebuild =
-                indexer->init(opCtx, collection, all, MultiIndexBlock::kNoopOnInitFn);
+            swIndexesToRebuild = indexer->init(
+                opCtx, collection, all, MultiIndexBlock::kNoopOnInitFn, /*forRecovery=*/false);
             uassertStatusOK(swIndexesToRebuild.getStatus());
             wunit.commit();
         });
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 9aaaa2b48c3..d94346896da 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -171,6 +171,7 @@ void makeUpdateRequest(OperationContext* opCtx,
     requestOut->setExplain(explain);
 
     requestOut->setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+    requestOut->setBypassEmptyTsReplacement(request.getBypassEmptyTsReplacement());
 }
 
 void makeDeleteRequest(OperationContext* opCtx,
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index 48dee5c1e7d..df96e09e2ce 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/client.h"
 #include "mongo/db/clientcursor.h"
+#include "mongo/db/collection_type.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/run_aggregate.h"
 #include "mongo/db/commands/test_commands_enabled.h"
@@ -54,6 +55,10 @@
 #include "mongo/db/query/find_common.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/query/query_knobs_gen.h"
+#include "mongo/db/query/query_shape/query_shape.h"
+#include "mongo/db/query/query_stats/find_key.h"
+#include "mongo/db/query/query_stats/key.h"
+#include "mongo/db/query/query_stats/query_stats.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/counters.h"
@@ -111,25 +116,6 @@ std::unique_ptr<FindCommandRequest> translateNtoReturnToLimitOrBatchSize(
     return findCmd;
 }
 
-// Parses the command object to a FindCommandRequest. If the client request did not specify any
-// runtime constants, make them available to the query here.
-std::unique_ptr<FindCommandRequest> parseCmdObjectToFindCommandRequest(OperationContext* opCtx,
-                                                                       NamespaceString nss,
-                                                                       BSONObj cmdObj) {
-    auto findCommand = query_request_helper::makeFromFindCommand(
-        std::move(cmdObj),
-        std::move(nss),
-        APIParameters::get(opCtx).getAPIStrict().value_or(false));
-
-    // Rewrite any FLE find payloads that exist in the query if this is a FLE 2 query.
-    if (shouldDoFLERewrite(findCommand)) {
-        invariant(findCommand->getNamespaceOrUUID().nss());
-        processFLEFindD(opCtx, findCommand->getNamespaceOrUUID().nss().get(), findCommand.get());
-    }
-
-    return translateNtoReturnToLimitOrBatchSize(std::move(findCommand));
-}
-
 boost::intrusive_ptr<ExpressionContext> makeExpressionContext(
     OperationContext* opCtx,
     const FindCommandRequest& findCommand,
@@ -145,38 +131,13 @@ boost::intrusive_ptr<ExpressionContext> makeExpressionContext(
         // ExpressionContext.
         collator = collPtr->getDefaultCollator()->clone();
     }
-
-    // Although both 'find' and 'aggregate' commands have an ExpressionContext, some of the data
-    // members in the ExpressionContext are used exclusively by the aggregation subsystem. This
-    // includes the following fields which here we simply initialize to some meaningless default
-    // value:
-    //  - explain
-    //  - fromMongos
-    //  - needsMerge
-    //  - bypassDocumentValidation
-    //  - mongoProcessInterface
-    //  - resolvedNamespaces
-    //  - uuid
-    //
-    // As we change the code to make the find and agg systems more tightly coupled, it would make
-    // sense to start initializing these fields for find operations as well.
-    auto expCtx = make_intrusive<ExpressionContext>(
-        opCtx,
-        verbosity,
-        false,  // fromMongos
-        false,  // needsMerge
-        findCommand.getAllowDiskUse().value_or(allowDiskUseByDefault.load()),
-        false,  // bypassDocumentValidation
-        false,  // isMapReduceCommand
-        findCommand.getNamespaceOrUUID().nss().value_or(NamespaceString()),
-        findCommand.getLegacyRuntimeConstants(),
-        std::move(collator),
-        nullptr,  // mongoProcessInterface
-        StringMap<ExpressionContext::ResolvedNamespace>{},
-        boost::none,                             // uuid
-        findCommand.getLet(),                    // let
-        CurOp::get(opCtx)->dbProfileLevel() > 0  // mayDbProfile
-    );
+    auto expCtx =
+        make_intrusive<ExpressionContext>(opCtx,
+                                          findCommand,
+                                          std::move(collator),
+                                          CurOp::get(opCtx)->dbProfileLevel() > 0,  // mayDbProfile
+                                          verbosity,
+                                          allowDiskUseByDefault.load());
     if (storageGlobalParams.readOnly) {
         // Disallow disk use if in read-only mode.
         expCtx->allowDiskUse = false;
@@ -198,6 +159,45 @@ void beginQueryOp(OperationContext* opCtx, const NamespaceString& nss, const BSO
 }
 
 /**
+ * Parses the grammar elements like 'filter', 'sort', and 'projection' from the raw
+ * 'FindCommandRequest', and tracks internal state like begining the operation's timer and recording
+ * query shape stats (if enabled).
+ */
+std::unique_ptr<CanonicalQuery> parseQueryAndBeginOperation(
+    OperationContext* opCtx,
+    const AutoGetCollectionForReadCommandMaybeLockFree& ctx,
+    const NamespaceString& nss,
+    BSONObj requestBody,
+    std::unique_ptr<FindCommandRequest> findCommand,
+    const CollectionPtr& collection) {
+    // Fill out curop information.
+    beginQueryOp(opCtx, nss, requestBody);
+    // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
+    const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
+
+    auto expCtx =
+        makeExpressionContext(opCtx, *findCommand, collection, boost::none /* verbosity */);
+
+    auto parsedRequest = uassertStatusOK(
+        parsed_find_command::parse(expCtx,
+                                   std::move(findCommand),
+                                   extensionsCallback,
+                                   MatchExpressionParser::kAllowAllSpecialFeatures));
+
+    // Register query stats collection. Exclude queries against collections with encrypted fields.
+    // It is important to do this before canonicalizing and optimizing the query, each of which
+    // would alter the query shape.
+    if (!(collection && collection.get()->getCollectionOptions().encryptedFieldConfig)) {
+        query_stats::registerRequest(opCtx, nss, [&]() {
+            return std::make_unique<query_stats::FindKey>(
+                expCtx, *parsedRequest, ctx.getCollectionType());
+        });
+    }
+
+    return uassertStatusOK(
+        CanonicalQuery::canonicalize(std::move(expCtx), std::move(parsedRequest)));
+}
+/**
  * A command for running .find() queries.
  */
 class FindCmd final : public Command {
@@ -322,7 +322,7 @@ public:
             const auto nss = ctx->getNss();
 
             // Parse the command BSON to a FindCommandRequest.
-            auto findCommand = parseCmdObjectToFindCommandRequest(opCtx, nss, _request.body);
+            auto findCommand = _parseCmdObjectToFindCommandRequest(opCtx, nss, _request.body);
 
             // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
             const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
@@ -372,8 +372,8 @@ public:
                 try {
                     // An empty PrivilegeVector is acceptable because these privileges are only
                     // checked on getMore and explain will not open a cursor.
-                    uassertStatusOK(runAggregate(
-                        opCtx, nss, aggRequest, viewAggCmd, PrivilegeVector(), result));
+                    uassertStatusOK(
+                        runAggregate(opCtx, aggRequest, viewAggCmd, PrivilegeVector(), result));
                 } catch (DBException& error) {
                     if (error.code() == ErrorCodes::InvalidPipelineOperator) {
                         uasserted(ErrorCodes::InvalidPipelineOperator,
@@ -419,10 +419,10 @@ public:
             // Parse the command BSON to a FindCommandRequest. Pass in the parsedNss in case cmdObj
             // does not have a UUID.
             auto parsedNss = NamespaceString{CommandHelpers::parseNsFromCommand(_dbName, cmdObj)};
-            const bool isExplain = false;
             const bool isOplogNss = (parsedNss == NamespaceString::kRsOplogNamespace);
             auto findCommand =
-                parseCmdObjectToFindCommandRequest(opCtx, std::move(parsedNss), cmdObj);
+                _parseCmdObjectToFindCommandRequest(opCtx, std::move(parsedNss), cmdObj);
+            CurOp::get(opCtx)->beginQueryPlanningTimer();
 
             // Only allow speculative majority for internal commands that specify the correct flag.
             uassert(ErrorCodes::ReadConcernMajorityNotEnabled,
@@ -542,21 +542,8 @@ public:
                     findCommand->getResumeAfter(), isClusteredCollection));
             }
 
-            // Fill out curop information.
-            beginQueryOp(opCtx, nss, _request.body);
-
-            // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
-            const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
-
-            auto expCtx =
-                makeExpressionContext(opCtx, *findCommand, collection, boost::none /* verbosity */);
-            auto cq = uassertStatusOK(
-                CanonicalQuery::canonicalize(opCtx,
-                                             std::move(findCommand),
-                                             isExplain,
-                                             std::move(expCtx),
-                                             extensionsCallback,
-                                             MatchExpressionParser::kAllowAllSpecialFeatures));
+            auto cq = parseQueryAndBeginOperation(
+                opCtx, *ctx, nss, _request.body, std::move(findCommand), collection);
 
             // If we are running a query against a view, or if we are trying to test the new
             // optimizer, redirect this query through the aggregation system.
@@ -573,6 +560,9 @@ public:
                 auto viewAggregationCommand =
                     uassertStatusOK(query_request_helper::asAggregationCommand(findCommand));
 
+                // This doesn't directly call 'runAggregate()' so it doesn't need to adapt to the
+                // new API on v6.0. @Alyssa this suggests we should look into view performance more
+                // carefully on v6.0. The perf of this code path may have different characteristics?
                 BSONObj aggResult = CommandHelpers::runCommandDirectly(
                     opCtx, OpMsgRequest::fromDBAndBody(_dbName, std::move(viewAggregationCommand)));
                 auto status = getStatusFromCommandResult(aggResult);
@@ -626,7 +616,7 @@ public:
                 // there is no ClientCursor id, and then return.
                 const long long numResults = 0;
                 const CursorId cursorId = 0;
-                endQueryOp(opCtx, collection, *exec, numResults, cursorId);
+                endQueryOp(opCtx, collection, *exec, numResults, boost::none, cmdObj);
                 auto bodyBuilder = result->getBodyBuilder();
                 appendCursorResponseObject(
                     cursorId, nss.ns(), BSONArray(), boost::none, &bodyBuilder);
@@ -725,11 +715,9 @@ public:
                     pinnedCursor.getCursor()->setLeftoverMaxTimeMicros(
                         opCtx->getRemainingMaxTimeMicros());
                 }
-                pinnedCursor.getCursor()->setNReturnedSoFar(numResults);
-                pinnedCursor.getCursor()->incNBatches();
 
                 // Fill out curop based on the results.
-                endQueryOp(opCtx, collection, *cursorExec, numResults, cursorId);
+                endQueryOp(opCtx, collection, *cursorExec, numResults, pinnedCursor, cmdObj);
 
                 if (stashResourcesForGetMore) {
                     // Collect storage stats now before we stash the recovery unit. These stats are
@@ -744,7 +732,7 @@ public:
                         opCtx->recoveryUnit()->computeOperationStatisticsSinceLastCall();
                 }
             } else {
-                endQueryOp(opCtx, collection, *exec, numResults, cursorId);
+                endQueryOp(opCtx, collection, *exec, numResults, boost::none, cmdObj);
             }
 
             // Generate the response object to send to the client.
@@ -785,6 +773,25 @@ public:
     private:
         const OpMsgRequest _request;
         const StringData _dbName;
+
+        // Parses the command object to a FindCommandRequest. If the client request did not specify
+        // any runtime constants, make them available to the query here.
+        std::unique_ptr<FindCommandRequest> _parseCmdObjectToFindCommandRequest(
+            OperationContext* opCtx, NamespaceString nss, BSONObj cmdObj) {
+            auto findCommand = query_request_helper::makeFromFindCommand(
+                std::move(cmdObj),
+                std::move(nss),
+                APIParameters::get(opCtx).getAPIStrict().value_or(false));
+
+            // Rewrite any FLE find payloads that exist in the query if this is a FLE 2 query.
+            if (shouldDoFLERewrite(findCommand)) {
+                invariant(findCommand->getNamespaceOrUUID().nss());
+                processFLEFindD(
+                    opCtx, findCommand->getNamespaceOrUUID().nss().value(), findCommand.get());
+            }
+
+            return translateNtoReturnToLimitOrBatchSize(std::move(findCommand));
+        }
     };
 
 } findCmd;
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index c699a3262e7..07df3bf5a9b 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -362,7 +362,7 @@ public:
          * be returned by this getMore.
          *
          * Returns true if the cursor should be saved for subsequent getMores, and false otherwise.
-         * Fills out *numResults with the number of documents in the batch, which must be
+         * Fills out 'numResults' with the number of documents in the batch, which must be
          * initialized to zero by the caller.
          *
          * Throws an exception on failure.
@@ -385,9 +385,14 @@ public:
             try {
                 while (!FindCommon::enoughForGetMore(batchSize, *numResults) &&
                        PlanExecutor::ADVANCED == (state = exec->getNext(&obj, nullptr))) {
+                    auto nextPostBatchResumeToken = exec->getPostBatchResumeToken();
+
                     // If adding this object will cause us to exceed the message size limit, then we
                     // stash it for later.
-                    if (!FindCommon::haveSpaceForNext(obj, *numResults, nextBatch->bytesUsed())) {
+                    if (!FindCommon::haveSpaceForNext(obj,
+                                                      *numResults,
+                                                      nextBatch->bytesUsed() +
+                                                          nextPostBatchResumeToken.objsize())) {
                         exec->stashResult(obj);
                         break;
                     }
@@ -396,7 +401,7 @@ public:
                     awaitDataState(opCtx).shouldWaitForInserts = false;
 
                     // If this executor produces a postBatchResumeToken, add it to the response.
-                    nextBatch->setPostBatchResumeToken(exec->getPostBatchResumeToken());
+                    nextBatch->setPostBatchResumeToken(nextPostBatchResumeToken);
 
                     // At this point, we know that there will be at least one document in this
                     // batch. Reserve an initial estimated number of bytes for the response.
@@ -719,12 +724,9 @@ public:
             // documents.
             auto& metricsCollector = ResourceConsumption::MetricsCollector::get(opCtx);
             metricsCollector.incrementDocUnitsReturned(docUnitsReturned);
-            cursorPin->incNReturnedSoFar(numResults);
-            cursorPin->incNBatches();
-
-            // Ensure log and profiler include the number of results returned in this getMore's
-            // response batch.
-            curOp->debug().nreturned = numResults;
+            curOp->debug().additiveMetrics.nBatches = 1;
+            curOp->setEndOfOpMetrics(numResults);
+            collectQueryStatsMongod(opCtx, cursorPin);
 
             if (respondWithId) {
                 cursorDeleter.dismiss();
diff --git a/src/mongo/db/commands/map_reduce_agg.cpp b/src/mongo/db/commands/map_reduce_agg.cpp
index eb1d432f5b5..91a27c93bf4 100644
--- a/src/mongo/db/commands/map_reduce_agg.cpp
+++ b/src/mongo/db/commands/map_reduce_agg.cpp
@@ -138,6 +138,8 @@ bool runAggregationMapReduce(OperationContext* opCtx,
     Timer cmdTimer;
 
     auto parsedMr = MapReduceCommandRequest::parse(IDLParserErrorContext("mapReduce"), cmd);
+    auto curop = CurOp::get(opCtx);
+    curop->beginQueryPlanningTimer();
     auto expCtx = makeExpressionContext(opCtx, parsedMr, verbosity);
     auto runnablePipeline = [&]() {
         auto pipeline = map_reduce_common::translateFromMR(parsedMr, expCtx);
@@ -146,10 +148,10 @@ bool runAggregationMapReduce(OperationContext* opCtx,
     }();
     auto exec = plan_executor_factory::make(expCtx, std::move(runnablePipeline));
     auto&& explainer = exec->getPlanExplainer();
-
+    // Store the plan summary string in CurOp.
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
-        CurOp::get(opCtx)->setPlanSummary_inlock(explainer.getPlanSummary());
+        curop->setPlanSummary_inlock(explainer.getPlanSummary());
     }
 
     try {
diff --git a/src/mongo/db/commands/map_reduce_agg_test.cpp b/src/mongo/db/commands/map_reduce_agg_test.cpp
index 52ee67c416b..b47b2bfa0e4 100644
--- a/src/mongo/db/commands/map_reduce_agg_test.cpp
+++ b/src/mongo/db/commands/map_reduce_agg_test.cpp
@@ -46,15 +46,6 @@
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/unittest/unittest.h"
 
-#define ASSERT_DOES_NOT_THROW(EXPRESSION)                                          \
-    try {                                                                          \
-        EXPRESSION;                                                                \
-    } catch (const AssertionException& e) {                                        \
-        str::stream err;                                                           \
-        err << "Threw an exception incorrectly: " << e.toString();                 \
-        ::mongo::unittest::TestAssertionFailure(__FILE__, __LINE__, err).stream(); \
-    }
-
 namespace mongo {
 namespace {
 
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index decfcb35fe8..2042914e16b 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -142,7 +142,6 @@ public:
                 opCtx, !Pipeline::aggHasWriteStage(_request.body));
 
             uassertStatusOK(runAggregate(opCtx,
-                                         _aggregationRequest.getNamespace(),
                                          _aggregationRequest,
                                          _liteParsedPipeline,
                                          _request.body,
@@ -165,7 +164,6 @@ public:
                      rpc::ReplyBuilderInterface* result) override {
 
             uassertStatusOK(runAggregate(opCtx,
-                                         _aggregationRequest.getNamespace(),
                                          _aggregationRequest,
                                          _liteParsedPipeline,
                                          _request.body,
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index b47645610f7..66d68f4413c 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -74,6 +74,9 @@
 #include "mongo/db/query/query_feature_flags_gen.h"
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/query/query_planner_common.h"
+#include "mongo/db/query/query_stats/agg_key.h"
+#include "mongo/db/query/query_stats/key.h"
+#include "mongo/db/query/query_stats/query_stats.h"
 #include "mongo/db/read_concern.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/read_concern_args.h"
@@ -627,7 +630,6 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx
 
         getSearchHelpers(expCtx->opCtx->getServiceContext())
             ->injectSearchShardFiltererIfNeeded(pipeline.get());
-
         // Complete creation of the initial $cursor stage, if needed.
         PipelineD::attachInnerQueryExecutorToPipeline(collections,
                                                       attachExecutorCallback.first,
@@ -640,7 +642,6 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx
             // There are separate ExpressionContexts for each exchange pipeline, so make sure to
             // pass the pipeline's ExpressionContext to the plan executor factory.
             auto pipelineExpCtx = pipelineIt->getContext();
-
             execs.emplace_back(
                 plan_executor_factory::make(std::move(pipelineExpCtx),
                                             std::move(pipelineIt),
@@ -664,15 +665,11 @@ Status runAggregateOnView(OperationContext* opCtx,
                           const MultipleCollectionAccessor& collections,
                           boost::optional<std::unique_ptr<CollatorInterface>> collatorToUse,
                           const ViewDefinition* view,
-                          const boost::intrusive_ptr<ExpressionContext>& expCtx,
                           std::shared_ptr<const CollectionCatalog> catalog,
                           const PrivilegeVector& privileges,
-                          CurOp* curOp,
                           rpc::ReplyBuilderInterface* result,
                           const std::function<void(void)>& resetContextFn) {
     auto nss = request.getNamespace();
-    checkCollectionUUIDMismatch(
-        opCtx, nss, collections.getMainCollection(), request.getCollectionUUID());
 
     uassert(ErrorCodes::CommandNotSupportedOnView,
             "mapReduce on a view is not supported",
@@ -719,7 +716,7 @@ Status runAggregateOnView(OperationContext* opCtx,
 
     auto status{Status::OK()};
     try {
-        status = runAggregate(opCtx, origNss, newRequest, newCmd, privileges, result);
+        status = runAggregate(opCtx, newRequest, newCmd, privileges, result, resolvedView, request);
     } catch (const ExceptionForCat<ErrorCategory::StaleShardVersionError>& ex) {
         // Since we expect the view to be UNSHARDED, if we reached to this point there are
         // two possibilities:
@@ -739,31 +736,145 @@ Status runAggregateOnView(OperationContext* opCtx,
         // Set the namespace of the curop back to the view namespace so ctx records
         // stats on this view namespace on destruction.
         stdx::lock_guard<Client> lk(*opCtx->getClient());
-        curOp->setNS_inlock(nss.ns());
+        CurOp::get(opCtx)->setNS_inlock(nss.ns());
     }
 
     return status;
 }
 
+/**
+ * Determines the collection type of the query by precedence of various configurations. The order
+ * of these checks is critical since there may be overlap (e.g., a view over a virtual collection
+ * is classified as a view).
+ */
+query_shape::CollectionType determineCollectionType(
+    const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx,
+    boost::optional<const ResolvedView&> resolvedView,
+    bool hasChangeStream,
+    bool isCollectionless) {
+    if (resolvedView.has_value()) {
+        if (resolvedView->timeseries()) {
+            return query_shape::CollectionType::kTimeseries;
+        }
+        return query_shape::CollectionType::kView;
+    }
+    if (isCollectionless) {
+        return query_shape::CollectionType::kVirtual;
+    }
+    if (hasChangeStream) {
+        return query_shape::CollectionType::kChangeStream;
+    }
+    return ctx ? ctx->getCollectionType() : query_shape::CollectionType::kUnknown;
+}
+
+std::unique_ptr<Pipeline, PipelineDeleter> parsePipelineAndRegisterQueryStats(
+    OperationContext* opCtx,
+    const NamespaceString& origNss,
+    const AggregateCommandRequest& request,
+    const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx,
+    std::unique_ptr<CollatorInterface> collator,
+    boost::optional<UUID> uuid,
+    ExpressionContext::CollationMatchesDefault collationMatchesDefault,
+    const MultipleCollectionAccessor& collections,
+    stdx::unordered_set<NamespaceString> pipelineInvolvedNamespaces,
+    const LiteParsedPipeline& liteParsedPipeline,
+    bool isCollectionless,
+    boost::optional<const ResolvedView&> resolvedView,
+    boost::optional<const AggregateCommandRequest&> origRequest) {
+    // If we're operating over a view, we first parse just the original user-given request
+    // for the sake of registering query stats. Then, we'll parse the view pipeline and stitch
+    // the two pipelines together below.
+    auto expCtx =
+        makeExpressionContext(opCtx, request, std::move(collator), uuid, collationMatchesDefault);
+    // If any involved collection contains extended-range data, set a flag which individual
+    // DocumentSource parsers can check.
+    collections.forEach([&](const CollectionPtr& coll) {
+        if (coll->getRequiresTimeseriesExtendedRangeSupport())
+            expCtx->setRequiresTimeseriesExtendedRangeSupport(true);
+    });
+
+    const bool hasChangeStream = liteParsedPipeline.hasChangeStream();
+    // A pipeline with $changeStreamSplitLargeEvent requires the use of resume token format
+    // v2, since the 'fragmentNum' field only exists in this version and later.
+    if (hasChangeStream && liteParsedPipeline.endsWithChangeStreamSplitLargeEvent()) {
+        expCtx->changeStreamTokenVersion = 2;
+    }
+
+    auto requestForQueryStats = origRequest.has_value() ? *origRequest : request;
+    expCtx->startExpressionCounters();
+    auto pipeline = Pipeline::parse(requestForQueryStats.getPipeline(), expCtx);
+    expCtx->stopExpressionCounters();
+
+    // Register query stats with the pre-optimized pipeline. Exclude queries against collections
+    // with encrypted fields. We still collect query stats on collection-less aggregations.
+    bool hasEncryptedFields = ctx && ctx->getCollection() &&
+        ctx->getCollection()->getCollectionOptions().encryptedFieldConfig;
+    if (!hasEncryptedFields) {
+        // If this is a query over a resolved view, we want to register query stats with the
+        // original user-given request and pipeline, rather than the new request generated when
+        // resolving the view.
+        auto collectionType =
+            determineCollectionType(ctx, resolvedView, hasChangeStream, isCollectionless);
+
+        query_stats::registerRequest(opCtx,
+                                     origNss,
+                                     [&]() {
+                                         return std::make_unique<query_stats::AggKey>(
+                                             requestForQueryStats,
+                                             *pipeline,
+                                             expCtx,
+                                             pipelineInvolvedNamespaces,
+                                             origNss,
+                                             collectionType);
+                                     },
+                                     hasChangeStream);
+    }
+
+    if (resolvedView.has_value()) {
+        expCtx->startExpressionCounters();
+
+        if (resolvedView->timeseries()) {
+            // For timeseries, there may have been rewrites done on the raw BSON pipeline
+            // during view resolution. We must parse the request's full resolved pipeline
+            // which will account for those rewrites.
+            // TODO SERVER-82101 Re-organize timeseries rewrites so timeseries can follow the
+            // same pattern here as other views
+            pipeline = Pipeline::parse(request.getPipeline(), expCtx);
+        } else {
+            // Parse the view pipeline, then stitch the user pipeline and view pipeline together
+            // to build the total aggregation pipeline.
+            auto userPipeline = std::move(pipeline);
+            pipeline = Pipeline::parse(resolvedView->getPipeline(), expCtx);
+            pipeline->appendPipeline(std::move(userPipeline));
+        }
+
+        expCtx->stopExpressionCounters();
+    }
+
+    return pipeline;
+}
 }  // namespace
 
 Status runAggregate(OperationContext* opCtx,
-                    const NamespaceString& nss,
                     AggregateCommandRequest& request,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result) {
-    return runAggregate(opCtx, nss, request, {request}, cmdObj, privileges, result);
+                    rpc::ReplyBuilderInterface* result,
+                    boost::optional<const ResolvedView&> resolvedView,
+                    boost::optional<const AggregateCommandRequest&> origRequest) {
+    return runAggregate(
+        opCtx, request, {request}, cmdObj, privileges, result, resolvedView, origRequest);
 }
 
 Status runAggregate(OperationContext* opCtx,
-                    const NamespaceString& origNss,
                     AggregateCommandRequest& request,
                     const LiteParsedPipeline& liteParsedPipeline,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result) {
-
+                    rpc::ReplyBuilderInterface* result,
+                    boost::optional<const ResolvedView&> resolvedView,
+                    boost::optional<const AggregateCommandRequest&> origRequest) {
+    auto origNss = origRequest.has_value() ? origRequest->getNamespace() : request.getNamespace();
     // Perform some validations on the LiteParsedPipeline and request before continuing with the
     // aggregation command.
     performValidationChecks(opCtx, request, liteParsedPipeline);
@@ -848,7 +959,6 @@ Status runAggregate(OperationContext* opCtx,
 
             // Raise an error if 'origNss' is a view. We do not need to check this if we are opening
             // a stream on an entire db or across the cluster.
-            const TenantDatabaseName origTenantDbName(boost::none, origNss.db());
             if (!origNss.isCollectionlessAggregateNS()) {
                 auto view = catalog->lookupView(opCtx, origNss);
                 uassert(ErrorCodes::CommandNotSupportedOnView,
@@ -882,7 +992,7 @@ Status runAggregate(OperationContext* opCtx,
                                  nss,
                                  Top::LockType::NotLocked,
                                  AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                 0);
+                                 catalog->getDatabaseProfileLevel(nss.db()));
             auto [collator, match] = PipelineD::resolveCollator(
                 opCtx, request.getCollation().get_value_or(BSONObj()), nullptr);
             collatorToUse.emplace(std::move(collator));
@@ -904,6 +1014,13 @@ Status runAggregate(OperationContext* opCtx,
             }
         }
 
+        // If collectionUUID was provided, verify the collection exists and has the expected UUID.
+        checkCollectionUUIDMismatch(opCtx,
+                                    nss,
+                                    collections.getMainCollection(),
+                                    request.getCollectionUUID(),
+                                    false /* checkFeatureFlag */);
+
         // If this is a view, resolve it by finding the underlying collection and stitching view
         // pipelines and this request's pipeline together. We then release our locks before
         // recursively calling runAggregate(), which will re-acquire locks on the underlying
@@ -921,41 +1038,29 @@ Status runAggregate(OperationContext* opCtx,
                                       collections,
                                       std::move(collatorToUse),
                                       ctx->getView(),
-                                      expCtx,
                                       catalog,
                                       privileges,
-                                      curOp,
                                       result,
                                       resetContext);
         }
 
-        // If collectionUUID was provided, verify the collection exists and has the expected UUID.
-        checkCollectionUUIDMismatch(opCtx,
-                                    nss,
-                                    collections.getMainCollection(),
-                                    request.getCollectionUUID(),
-                                    false /* checkFeatureFlag */);
-
         invariant(collatorToUse);
-        expCtx = makeExpressionContext(
-            opCtx, request, std::move(*collatorToUse), uuid, collatorToUseMatchesDefault);
-
-        // If any involved collection contains extended-range data, set a flag which individual
-        // DocumentSource parsers can check.
-        collections.forEach([&](const CollectionPtr& coll) {
-            if (coll->getRequiresTimeseriesExtendedRangeSupport())
-                expCtx->setRequiresTimeseriesExtendedRangeSupport(true);
-        });
-
-        // A pipeline with $changeStreamSplitLargeEvent requires the use of resume token format v2,
-        // since the 'fragmentNum' field only exists in this version and later.
-        if (hasChangeStream && liteParsedPipeline.endsWithChangeStreamSplitLargeEvent()) {
-            expCtx->changeStreamTokenVersion = 2;
-        }
-
-        expCtx->startExpressionCounters();
-        auto pipeline = Pipeline::parse(request.getPipeline(), expCtx);
-        expCtx->stopExpressionCounters();
+        auto pipeline = parsePipelineAndRegisterQueryStats(opCtx,
+                                                           origNss,
+                                                           request,
+                                                           ctx,
+                                                           std::move(*collatorToUse),
+                                                           uuid,
+                                                           collatorToUseMatchesDefault,
+                                                           collections,
+                                                           pipelineInvolvedNamespaces,
+                                                           liteParsedPipeline,
+                                                           nss.isCollectionlessAggregateNS(),
+                                                           resolvedView,
+                                                           origRequest);
+        expCtx = pipeline->getContext();
+
+        CurOp::get(opCtx)->beginQueryPlanningTimer();
 
         if (!request.getAllowDiskUse().value_or(true)) {
             allowDiskUseFalseCounter.increment();
@@ -1034,6 +1139,7 @@ Status runAggregate(OperationContext* opCtx,
         }
     });
     for (auto&& exec : execs) {
+        // TODO SERVER-79373: Do not create a cursor if results can fit in a single batch.
         ClientCursorParams cursorParams(
             std::move(exec),
             origNss,
@@ -1077,6 +1183,7 @@ Status runAggregate(OperationContext* opCtx,
                                    cmdObj,
                                    &bodyBuilder);
         }
+        collectQueryStatsMongod(opCtx, std::move(curOp->debug().queryStatsInfo.key));
     } else {
         // Cursor must be specified, if explain is not.
         const bool keepCursor = handleCursorCommand(
@@ -1089,13 +1196,15 @@ Status runAggregate(OperationContext* opCtx,
         PlanSummaryStats stats;
         planExplainer.getSummaryStats(&stats);
         curOp->debug().setPlanSummaryMetrics(stats);
-        curOp->debug().nreturned = stats.nReturned;
+        curOp->setEndOfOpMetrics(stats.nReturned);
+
+        collectQueryStatsMongod(opCtx, pins[0]);
 
-        // For an optimized away pipeline, signal the cache that a query operation has completed.
-        // For normal pipelines this is done in DocumentSourceCursor.
+        // For an optimized away pipeline, signal the cache that a query operation has
+        // completed. For normal pipelines this is done in DocumentSourceCursor.
         if (ctx) {
-            // Due to yielding, the collection pointers saved in MultipleCollectionAccessor might
-            // have become invalid. We will need to refresh them here.
+            // Due to yielding, the collection pointers saved in MultipleCollectionAccessor
+            // might have become invalid. We will need to refresh them here.
             collections = MultipleCollectionAccessor(opCtx,
                                                      &ctx->getCollection(),
                                                      ctx->getNss(),
@@ -1118,10 +1227,11 @@ Status runAggregate(OperationContext* opCtx,
         }
     }
 
-    // The aggregation pipeline may change the namespace of the curop and we need to set it back to
-    // the original namespace to correctly report command stats. One example when the namespace can
-    // be changed is when the pipeline contains an $out stage, which executes an internal command to
-    // create a temp collection, changing the curop namespace to the name of this temp collection.
+    // The aggregation pipeline may change the namespace of the curop and we need to set it back
+    // to the original namespace to correctly report command stats. One example when the
+    // namespace can be changed is when the pipeline contains an $out stage, which executes an
+    // internal command to create a temp collection, changing the curop namespace to the name of
+    // this temp collection.
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
         curOp->setNS_inlock(origNss.ns());
@@ -1129,5 +1239,4 @@ Status runAggregate(OperationContext* opCtx,
 
     return Status::OK();
 }
-
 }  // namespace mongo
diff --git a/src/mongo/db/commands/run_aggregate.h b/src/mongo/db/commands/run_aggregate.h
index b61538fb93d..ba73245bdbf 100644
--- a/src/mongo/db/commands/run_aggregate.h
+++ b/src/mongo/db/commands/run_aggregate.h
@@ -49,25 +49,30 @@ namespace mongo {
  * 'privileges' contains the privileges that were required to run this aggregation, to be used later
  * for re-checking privileges for getMore commands.
  *
+ * If the query over a view that's already been resolved, the resolved view and the original
+ * user-provided request both must be provided.
+ *
  * On success, fills out 'result' with the command response.
  */
 Status runAggregate(OperationContext* opCtx,
-                    const NamespaceString& nss,
                     AggregateCommandRequest& request,
                     const LiteParsedPipeline& liteParsedPipeline,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result);
+                    rpc::ReplyBuilderInterface* result,
+                    boost::optional<const ResolvedView&> resolvedView = boost::none,
+                    boost::optional<const AggregateCommandRequest&> origRequest = boost::none);
 
 /**
  * Convenience version that internally constructs the LiteParsedPipeline.
  */
 Status runAggregate(OperationContext* opCtx,
-                    const NamespaceString& nss,
                     AggregateCommandRequest& request,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result);
+                    rpc::ReplyBuilderInterface* result,
+                    boost::optional<const ResolvedView&> resolvedView = boost::none,
+                    boost::optional<const AggregateCommandRequest&> origRequest = boost::none);
 
 /**
  * Tracks explicit use of allowDiskUse:false with find and aggregate commands.
diff --git a/src/mongo/db/commands/server_status_metric.h b/src/mongo/db/commands/server_status_metric.h
index ff546527a8d..423c49e2f5f 100644
--- a/src/mongo/db/commands/server_status_metric.h
+++ b/src/mongo/db/commands/server_status_metric.h
@@ -32,8 +32,13 @@
 #include <string>
 
 #include "mongo/db/jsobj.h"
+#include "mongo/platform/atomic_word.h"
 
 namespace mongo {
+class Atomic64Metric;
+
+template <>
+struct BSONObjAppendFormat<Atomic64Metric> : FormatKind<NumberLong> {};
 
 class ServerStatusMetric {
 public:
@@ -84,4 +89,36 @@ public:
 private:
     const T* _t;
 };
+
+/**
+ * Atomic wrapper for long long type for Metrics.  This is for values which are set rather than
+ * just incremented or decremented; if you want a counter, use Counter64.
+ */
+class Atomic64Metric {
+public:
+    /** Set _value to the max of the current or newMax. */
+    void setIfMax(long long newMax) {
+        /*  Note: compareAndSwap will load into val most recent value. */
+        for (long long val = _value.load(); val < newMax && !_value.compareAndSwap(&val, newMax);) {
+        }
+    }
+
+    /** store val into value. */
+    void set(long long val) {
+        _value.storeRelaxed(val);
+    }
+
+    /** Return the current value. */
+    long long get() const {
+        return _value.loadRelaxed();
+    }
+
+    /** TODO: SERVER-73806 Avoid implicit conversion to long long */
+    operator long long() const {
+        return get();
+    }
+
+private:
+    mongo::AtomicWord<long long> _value;
+};
 }  // namespace mongo
diff --git a/src/mongo/db/commands/set_cluster_parameter_invocation.cpp b/src/mongo/db/commands/set_cluster_parameter_invocation.cpp
index 6d32f73b393..b19b01ead17 100644
--- a/src/mongo/db/commands/set_cluster_parameter_invocation.cpp
+++ b/src/mongo/db/commands/set_cluster_parameter_invocation.cpp
@@ -126,6 +126,11 @@ StatusWith<bool> ClusterParameterDBClientService::updateParameterOnDisk(
         return Status(ErrorCodes::FailedToParse, errmsg);
     }
 
+    auto responseStatus = response.toStatus();
+    if (!responseStatus.isOK()) {
+        return responseStatus;
+    }
+
     return response.getNModified() > 0 || response.getN() > 0;
 }
 
diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
index 2c74781c36e..7462b287ed8 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -631,14 +631,6 @@ private:
                     tenantDbName,
                     MODE_X,
                     [&](const CollectionPtr& collection) {
-                        if (collection->getTimeseriesBucketsMayHaveMixedSchemaData()) {
-                            // The catalog entry flag has already been added. This can happen if the
-                            // upgrade process was interrupted and is being run again, or if there
-                            // was a time-series collection created during the upgrade. The upgrade
-                            // process cannot be aborted at this point.
-                            return true;
-                        }
-
                         NamespaceStringOrUUID nsOrUUID(dbName, collection->uuid());
                         CollMod collModCmd(collection->ns());
                         BSONObjBuilder unusedBuilder;
diff --git a/src/mongo/db/commands/user_management_commands.cpp b/src/mongo/db/commands/user_management_commands.cpp
index 13c5650bc98..ff0af0d0f0d 100644
--- a/src/mongo/db/commands/user_management_commands.cpp
+++ b/src/mongo/db/commands/user_management_commands.cpp
@@ -1450,7 +1450,6 @@ UsersInfoReply CmdUMCTyped<UsersInfoCommand, UMCInfoParams>::Invocation::typedRu
                                            std::move(pipeline));
         // Impose no cursor privilege requirements, as cursor is drained internally
         uassertStatusOK(runAggregate(opCtx,
-                                     AuthorizationManager::usersCollectionNamespace,
                                      aggRequest,
                                      aggregation_request_helper::serializeToCommandObj(aggRequest),
                                      PrivilegeVector(),
diff --git a/src/mongo/db/commands/validate.cpp b/src/mongo/db/commands/validate.cpp
index c17184cb14a..1ca15aca557 100644
--- a/src/mongo/db/commands/validate.cpp
+++ b/src/mongo/db/commands/validate.cpp
@@ -31,10 +31,22 @@
 
 #include "mongo/platform/basic.h"
 
+
+#include "mongo/base/error_codes.h"
+#include "mongo/base/status.h"
+#include "mongo/base/string_data.h"
+#include "mongo/bson/bson_validate_gen.h"
+#include "mongo/bson/bsonelement.h"
+#include "mongo/bson/bsonmisc.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/auth/action_type.h"
+#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_validation.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/commands/test_commands_enabled.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/storage/record_store.h"
@@ -331,7 +343,9 @@ public:
         }
 
         CollectionValidation::AdditionalOptions additionalOptions;
-        additionalOptions.warnOnSchemaValidation = cmdObj["warnOnSchemaValidation"].trueValue();
+        additionalOptions.validationVersion = getTestCommandsEnabled()
+            ? (ValidationVersion)bsonTestValidationVersion
+            : currentValidationVersion;
 
         ValidateResults validateResults;
         Status status = CollectionValidation::validate(opCtx,
diff --git a/src/mongo/db/commands/write_commands.cpp b/src/mongo/db/commands/write_commands.cpp
index bc031c90239..7196c905f1b 100644
--- a/src/mongo/db/commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands.cpp
@@ -1676,6 +1676,7 @@ public:
             updateRequest.setLegacyRuntimeConstants(request().getLegacyRuntimeConstants().value_or(
                 Variables::generateRuntimeConstants(opCtx)));
             updateRequest.setLetParameters(request().getLet());
+            updateRequest.setBypassEmptyTsReplacement(request().getBypassEmptyTsReplacement());
             updateRequest.setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
             updateRequest.setExplain(verbosity);