Update upstream source from tag 'upstream/6.0.0'master

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

1 files changed, 36 insertions, 136 deletions

diff --git a/src/mongo/s/query/cluster_aggregate.cpp b/src/mongo/s/query/cluster_aggregate.cpp
index 6374bcfd494..b3df71582da 100644
--- a/src/mongo/s/query/cluster_aggregate.cpp
+++ b/src/mongo/s/query/cluster_aggregate.cpp
@@ -27,7 +27,6 @@
  *    it in the license file.
  */
 
-#include "mongo/s/chunk_manager.h"
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
 
 #include "mongo/platform/basic.h"
@@ -38,7 +37,6 @@
 
 #include "mongo/db/api_parameters.h"
 #include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/catalog/collection_uuid_mismatch_info.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/curop.h"
@@ -57,9 +55,6 @@
 #include "mongo/db/query/explain_common.h"
 #include "mongo/db/query/find_common.h"
 #include "mongo/db/query/fle/server_rewrite.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/timeseries/timeseries_options.h"
 #include "mongo/db/views/resolved_view.h"
 #include "mongo/db/views/view.h"
@@ -99,7 +94,7 @@ namespace {
 // definition. It's okay that this is incorrect, we will repopulate the real namespace map on the
 // mongod. Note that this function must be called before forwarding an aggregation command on an
 // unsharded collection, in order to verify that the involved namespaces are allowed to be sharded.
-auto resolveInvolvedNamespaces(const stdx::unordered_set<NamespaceString>& involvedNamespaces) {
+auto resolveInvolvedNamespaces(stdx::unordered_set<NamespaceString> involvedNamespaces) {
     StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
     for (auto&& nss : involvedNamespaces) {
         resolvedNamespaces.try_emplace(nss.coll(), nss, std::vector<BSONObj>{});
@@ -262,68 +257,6 @@ std::vector<BSONObj> rebuildPipelineWithTimeSeriesGranularity(const std::vector<
     return newPipeline;
 }
 
-/**
- * Builds an expCtx with which to parse the request's pipeline, then parses the pipeline and
- * registers the pre-optimized pipeline with query stats collection.
- */
-std::unique_ptr<Pipeline, PipelineDeleter> parsePipelineAndRegisterQueryStats(
-    OperationContext* opCtx,
-    const stdx::unordered_set<NamespaceString>& involvedNamespaces,
-    const NamespaceString& executionNss,
-    AggregateCommandRequest& request,
-    const boost::optional<ChunkManager>& cm,
-    const LiteParsedPipeline& liteParsedPipeline,
-    bool hasChangeStream,
-    bool shouldDoFLERewrite) {
-    // Populate the collection UUID and the appropriate collation to use.
-    auto [collationObj, uuid] = [&]() -> std::pair<BSONObj, boost::optional<UUID>> {
-        // If this is a change stream, take the user-defined collation if one exists, or an
-        // empty BSONObj otherwise. Change streams never inherit the collection's default
-        // collation, and since collectionless aggregations generally run on the 'admin'
-        // database, the standard logic would attempt to resolve its non-existent UUID and
-        // collation by sending a specious 'listCollections' command to the config servers.
-        if (hasChangeStream) {
-            return {request.getCollation().value_or(BSONObj()), boost::none};
-        }
-
-        return cluster_aggregation_planner::getCollationAndUUID(
-            opCtx, cm, executionNss, request.getCollation().value_or(BSONObj()));
-    }();
-
-    // Build an ExpressionContext for the pipeline. This instantiates an appropriate collator,
-    // resolves all involved namespaces, and creates a shared MongoProcessInterface for use by the
-    // pipeline's stages.
-    boost::intrusive_ptr<ExpressionContext> expCtx =
-        makeExpressionContext(opCtx,
-                              request,
-                              collationObj,
-                              uuid,
-                              resolveInvolvedNamespaces(involvedNamespaces),
-                              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;
-    }
-
-    // Parse and optimize the full pipeline.
-    auto pipeline = Pipeline::parse(request.getPipeline(), expCtx);
-
-    // Skip query stats recording for queryable encryption queries.
-    if (!shouldDoFLERewrite) {
-        query_stats::registerRequest(
-            opCtx,
-            executionNss,
-            [&]() {
-                return std::make_unique<query_stats::AggKey>(
-                    request, *pipeline, expCtx, involvedNamespaces, executionNss);
-            },
-            hasChangeStream);
-    }
-    return pipeline;
-}
-
 }  // namespace
 
 Status ClusterAggregate::runAggregate(OperationContext* opCtx,
@@ -376,7 +309,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
     auto hasChangeStream = liteParsedPipeline.hasChangeStream();
     auto involvedNamespaces = liteParsedPipeline.getInvolvedNamespaces();
     auto shouldDoFLERewrite = ::mongo::shouldDoFLERewrite(request);
-    auto startsWithDocuments = liteParsedPipeline.startsWithDocuments();
 
     // If the routing table is not already taken by the higher level, fill it now.
     if (!cm) {
@@ -391,14 +323,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
             sharded_agg_helpers::getExecutionNsRoutingInfo(opCtx, namespaces.executionNss);
 
         if (!executionNsRoutingInfoStatus.isOK()) {
-            uassert(CollectionUUIDMismatchInfo(request.getDbName().toString(),
-                                               *request.getCollectionUUID(),
-                                               request.getNamespace().coll().toString(),
-                                               boost::none),
-                    "Database does not exist",
-                    executionNsRoutingInfoStatus != ErrorCodes::NamespaceNotFound ||
-                        !request.getCollectionUUID());
-
             if (liteParsedPipeline.startsWithCollStats()) {
                 uassertStatusOKWithContext(executionNsRoutingInfoStatus,
                                            "Unable to retrieve information for $collStats stage");
@@ -407,7 +331,7 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
 
         if (executionNsRoutingInfoStatus.isOK()) {
             cm = std::move(executionNsRoutingInfoStatus.getValue());
-        } else if (!((hasChangeStream || startsWithDocuments) &&
+        } else if (!(hasChangeStream &&
                      executionNsRoutingInfoStatus == ErrorCodes::NamespaceNotFound)) {
             appendEmptyResultSetWithStatus(
                 opCtx, namespaces.requestedNss, executionNsRoutingInfoStatus.getStatus(), result);
@@ -417,15 +341,33 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
 
     boost::intrusive_ptr<ExpressionContext> expCtx;
     const auto pipelineBuilder = [&]() {
-        auto pipeline = parsePipelineAndRegisterQueryStats(opCtx,
-                                                           involvedNamespaces,
-                                                           namespaces.executionNss,
-                                                           request,
-                                                           cm,
-                                                           liteParsedPipeline,
-                                                           hasChangeStream,
-                                                           shouldDoFLERewrite);
-        expCtx = pipeline->getContext();
+        // Populate the collection UUID and the appropriate collation to use.
+        auto [collationObj, uuid] = [&]() -> std::pair<BSONObj, boost::optional<UUID>> {
+            // If this is a change stream, take the user-defined collation if one exists, or an
+            // empty BSONObj otherwise. Change streams never inherit the collection's default
+            // collation, and since collectionless aggregations generally run on the 'admin'
+            // database, the standard logic would attempt to resolve its non-existent UUID and
+            // collation by sending a specious 'listCollections' command to the config servers.
+            if (hasChangeStream) {
+                return {request.getCollation().value_or(BSONObj()), boost::none};
+            }
+
+            return cluster_aggregation_planner::getCollationAndUUID(
+                opCtx, cm, namespaces.executionNss, request.getCollation().value_or(BSONObj()));
+        }();
+
+        // Build an ExpressionContext for the pipeline. This instantiates an appropriate collator,
+        // resolves all involved namespaces, and creates a shared MongoProcessInterface for use by
+        // the pipeline's stages.
+        expCtx = makeExpressionContext(opCtx,
+                                       request,
+                                       collationObj,
+                                       uuid,
+                                       resolveInvolvedNamespaces(involvedNamespaces),
+                                       hasChangeStream);
+
+        // Parse and optimize the full pipeline.
+        auto pipeline = Pipeline::parse(request.getPipeline(), expCtx);
 
         // If the aggregate command supports encrypted collections, do rewrites of the pipeline to
         // support querying against encrypted fields.
@@ -439,11 +381,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
         }
 
         pipeline->optimizePipeline();
-
-        // Validate the pipeline post-optimization.
-        const bool alreadyOptimized = true;
-        pipeline->validateCommon(alreadyOptimized);
-
         return pipeline;
     };
 
@@ -458,7 +395,6 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
         cm,
         involvedNamespaces,
         hasChangeStream,
-        startsWithDocuments,
         allowedToPassthrough,
         request.getPassthroughToShard().has_value());
 
@@ -471,48 +407,15 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
                 cluster_aggregation_planner::AggregationTargeter::TargetingPolicy::kMongosRequired);
 
     if (!expCtx) {
-        // When the AggregationTargeter chooses a "passthrough" or "specific shard only" policy, it
-        // does not call the 'pipelineBuilder' function, so we've yet to construct an expression
-        // context or register query stats. Because this is a passthrough, we only need a bare
-        // minimum expression context on mongos.
+        // When the AggregationTargeter chooses a "passthrough" policy, it does not call the
+        // 'pipelineBuilder' function, so we never get an expression context. Because this is a
+        // passthrough, we only need a bare minimum expression context anyway.
         invariant(targeter.policy ==
                       cluster_aggregation_planner::AggregationTargeter::kPassthrough ||
                   targeter.policy ==
                       cluster_aggregation_planner::AggregationTargeter::kSpecificShardOnly);
-
         expCtx = make_intrusive<ExpressionContext>(
             opCtx, nullptr, namespaces.executionNss, boost::none, request.getLet());
-        expCtx->addResolvedNamespaces(involvedNamespaces);
-
-
-        // We might need 'inMongos' temporarily set to true for query stats parsing, but we don't
-        // want to modify the value of 'expCtx' for future code execution so we will set it back to
-        // its original value.
-        ON_BLOCK_EXIT([&expCtx, originalInMongosVal = expCtx->inMongos]() {
-            expCtx->inMongos = originalInMongosVal;
-        });
-
-        // In order to parse a change stream request for query stats, 'inMongos' needs
-        // to be set to true.
-        if (hasChangeStream) {
-            expCtx->inMongos = true;
-        }
-
-        // Skip query stats recording for queryable encryption queries.
-        if (!shouldDoFLERewrite) {
-            // We want to hold off parsing the pipeline until it's clear we must. Because of that,
-            // we wait to parse the pipeline until this callback is invoked within
-            // query_stats::registerRequest.
-            query_stats::registerRequest(
-                opCtx,
-                namespaces.executionNss,
-                [&]() {
-                    auto pipeline = Pipeline::parse(request.getPipeline(), expCtx);
-                    return std::make_unique<query_stats::AggKey>(
-                        request, *pipeline, expCtx, involvedNamespaces, namespaces.executionNss);
-                },
-                hasChangeStream);
-        }
     }
 
     if (request.getExplain()) {
@@ -540,11 +443,10 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
                 // If this is an explain write the explain output and return.
                 auto expCtx = targeter.pipeline->getContext();
                 if (expCtx->explain) {
-                    auto opts = SerializationOptions{};
-                    opts.verbosity = boost::make_optional(ExplainOptions::Verbosity::kQueryPlanner);
                     *result << "splitPipeline" << BSONNULL << "mongos"
                             << Document{{"host", getHostNameCachedAndPort()},
-                                        {"stages", targeter.pipeline->writeExplainOps(opts)}};
+                                        {"stages",
+                                         targeter.pipeline->writeExplainOps(*expCtx->explain)}};
                     return Status::OK();
                 }
 
@@ -567,8 +469,7 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
                     namespaces,
                     privileges,
                     result,
-                    hasChangeStream,
-                    startsWithDocuments);
+                    hasChangeStream);
             }
             case cluster_aggregation_planner::AggregationTargeter::TargetingPolicy::
                 kSpecificShardOnly: {
@@ -613,12 +514,11 @@ Status ClusterAggregate::runAggregate(OperationContext* opCtx,
         updateHostsTargetedMetrics(opCtx, namespaces.executionNss, cm, involvedNamespaces);
         // Report usage statistics for each stage in the pipeline.
         liteParsedPipeline.tickGlobalStageCounters();
+
         // Add 'command' object to explain output.
         if (expCtx->explain) {
             explain_common::appendIfRoom(
                 aggregation_request_helper::serializeToCommandObj(request), "command", result);
-            collectQueryStatsMongos(opCtx,
-                                    std::move(CurOp::get(opCtx)->debug().queryStatsInfo.key));
         }
     }
     return status;