diff options
Diffstat (limited to 'src/mongo/db/pipeline/document_source_change_stream.cpp')
| -rw-r--r-- | src/mongo/db/pipeline/document_source_change_stream.cpp | 63 |
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, |