Update upstream source from tag 'upstream/6.0.0'master

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

1 files changed, 32 insertions, 31 deletions

diff --git a/src/mongo/db/pipeline/document_source_change_stream.cpp b/src/mongo/db/pipeline/document_source_change_stream.cpp
index 3f19d4ac4ee..4fc0041cc0e 100644
--- a/src/mongo/db/pipeline/document_source_change_stream.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream.cpp
@@ -37,7 +37,6 @@
 #include "mongo/db/pipeline/aggregate_command_gen.h"
 #include "mongo/db/pipeline/change_stream_constants.h"
 #include "mongo/db/pipeline/change_stream_filter_helpers.h"
-#include "mongo/db/pipeline/change_stream_helpers.h"
 #include "mongo/db/pipeline/change_stream_helpers_legacy.h"
 #include "mongo/db/pipeline/document_path_support.h"
 #include "mongo/db/pipeline/document_source_change_stream_add_post_image.h"
@@ -48,7 +47,6 @@
 #include "mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h"
 #include "mongo/db/pipeline/document_source_change_stream_handle_topology_change.h"
 #include "mongo/db/pipeline/document_source_change_stream_oplog_match.h"
-#include "mongo/db/pipeline/document_source_change_stream_split_large_event.h"
 #include "mongo/db/pipeline/document_source_change_stream_transform.h"
 #include "mongo/db/pipeline/document_source_change_stream_unwind_transaction.h"
 #include "mongo/db/pipeline/document_source_limit.h"
@@ -228,6 +226,23 @@ std::string DocumentSourceChangeStream::regexEscapeNsForChangeStream(StringData
     return result;
 }
 
+ResumeTokenData DocumentSourceChangeStream::resolveResumeTokenFromSpec(
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    const DocumentSourceChangeStreamSpec& spec) {
+    if (spec.getStartAfter()) {
+        return spec.getStartAfter()->getData();
+    } else if (spec.getResumeAfter()) {
+        return spec.getResumeAfter()->getData();
+    } else if (spec.getStartAtOperationTime()) {
+        return ResumeToken::makeHighWaterMarkToken(*spec.getStartAtOperationTime(),
+                                                   expCtx->changeStreamTokenVersion)
+            .getData();
+    }
+    tasserted(5666901,
+              "Expected one of 'startAfter', 'resumeAfter' or 'startAtOperationTime' to be "
+              "populated in $changeStream spec");
+}
+
 Timestamp DocumentSourceChangeStream::getStartTimeForNewStream(
     const boost::intrusive_ptr<ExpressionContext>& expCtx) {
     // If we do not have an explicit starting point, we should start from the latest majority
@@ -257,30 +272,6 @@ list<intrusive_ptr<DocumentSource>> DocumentSourceChangeStream::createFromBson(
     // Make sure that it is legal to run this $changeStream before proceeding.
     DocumentSourceChangeStream::assertIsLegalSpecification(expCtx, spec);
 
-    // If the user did not specify an explicit starting point, set it to the current time.
-    if (!spec.getResumeAfter() && !spec.getStartAfter() && !spec.getStartAtOperationTime()) {
-        // Make sure we update the 'startAtOperationTime' in the 'spec' so that we serialize the
-        // correct start point when sending it to the shards.
-        spec.setStartAtOperationTime(DocumentSourceChangeStream::getStartTimeForNewStream(expCtx));
-    }
-
-    // If the stream's default version differs from the client's token version, adopt the higher.
-    // This is the token version that will be used once the stream has passed the resume token.
-    const auto clientToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
-    expCtx->changeStreamTokenVersion =
-        std::max(expCtx->changeStreamTokenVersion, clientToken.version);
-
-    // If the user explicitly requested to resume from a high water mark token, but its version
-    // differs from the version chosen above, regenerate it with the new version. There is no need
-    // for a resumed HWM stream to adopt the old token version for events at the same clusterTime.
-    const bool tokenVersionsDiffer = (clientToken.version != expCtx->changeStreamTokenVersion);
-    const bool isHighWaterMark = ResumeToken::isHighWaterMarkToken(clientToken);
-    if (isHighWaterMark && tokenVersionsDiffer && (spec.getResumeAfter() || spec.getStartAfter())) {
-        spec.setResumeAfter(ResumeToken(ResumeToken::makeHighWaterMarkToken(
-            clientToken.clusterTime, expCtx->changeStreamTokenVersion)));
-        spec.setStartAfter(boost::none);
-    }
-
     // Save a copy of the spec on the expression context. Used when building the oplog filter.
     expCtx->changeStreamSpec = spec;
 
@@ -296,8 +287,15 @@ std::list<boost::intrusive_ptr<DocumentSource>> DocumentSourceChangeStream::_bui
     const boost::intrusive_ptr<ExpressionContext>& expCtx, DocumentSourceChangeStreamSpec spec) {
     std::list<boost::intrusive_ptr<DocumentSource>> stages;
 
+    // If the user did not specify an explicit starting point, set it to the current time.
+    if (!spec.getResumeAfter() && !spec.getStartAfter() && !spec.getStartAtOperationTime()) {
+        // Make sure we update the 'startAtOperationTime' in the 'spec' so that we serialize the
+        // correct start point when sending it to the shards.
+        spec.setStartAtOperationTime(DocumentSourceChangeStream::getStartTimeForNewStream(expCtx));
+    }
+
     // Obtain the resume token from the spec. This will be used when building the pipeline.
-    auto resumeToken = change_stream::resolveResumeTokenFromSpec(expCtx, spec);
+    auto resumeToken = DocumentSourceChangeStream::resolveResumeTokenFromSpec(expCtx, spec);
 
     // Unfold the $changeStream into its constituent stages and add them to the pipeline.
     stages.push_back(DocumentSourceChangeStreamOplogMatch::create(expCtx, spec));
@@ -312,9 +310,11 @@ std::list<boost::intrusive_ptr<DocumentSource>> DocumentSourceChangeStream::_bui
     // whether the event that matches the resume token should be followed by an "invalidate" event.
     stages.push_back(DocumentSourceChangeStreamCheckInvalidate::create(expCtx, spec));
 
-    // Always include a DSCSCheckResumability stage, both to verify that there is enough history to
-    // cover the change stream's starting point, and to swallow all events up to the resume point.
-    stages.push_back(DocumentSourceChangeStreamCheckResumability::create(expCtx, spec));
+    // If the starting point is a high water mark, or if we will be splitting the pipeline for
+    // dispatch to the shards in a cluster, we must include a DSCSCheckResumability stage.
+    if (expCtx->inMongos || ResumeToken::isHighWaterMarkToken(resumeToken)) {
+        stages.push_back(DocumentSourceChangeStreamCheckResumability::create(expCtx, spec));
+    }
 
     // If the pipeline is built on MongoS, we check for topology change events here. If a topology
     // change event is detected, this stage forwards the event directly to the executor via an
@@ -324,6 +324,7 @@ std::list<boost::intrusive_ptr<DocumentSource>> DocumentSourceChangeStream::_bui
         stages.push_back(DocumentSourceChangeStreamCheckTopologyChange::create(expCtx));
     }
 
+
     // If 'fullDocumentBeforeChange' is not set to 'off', add the DSCSAddPreImage stage into the
     // pipeline. We place this stage here so that any $match stages which follow the $changeStream
     // pipeline may be able to skip ahead of the DSCSAddPreImage stage. This allows a whole-db or
@@ -443,7 +444,7 @@ void DocumentSourceChangeStream::assertIsLegalSpecification(
             !spec.getResumeAfter() || !spec.getStartAfter());
 
     auto resumeToken = (spec.getResumeAfter() || spec.getStartAfter())
-        ? change_stream::resolveResumeTokenFromSpec(expCtx, spec)
+        ? resolveResumeTokenFromSpec(expCtx, spec)
         : boost::optional<ResumeTokenData>();
 
     uassert(40674,