diff options
21 files changed, 747 insertions, 54 deletions
diff --git a/jstests/change_streams/collation.js b/jstests/change_streams/collation.js index 4abc2f06ffb..531690afd19 100644 --- a/jstests/change_streams/collation.js +++ b/jstests/change_streams/collation.js @@ -240,9 +240,9 @@ assertDropCollection(db, collName); // Test that a $changeStream is allowed to resume on the dropped collection with an explicit // collation, even if it doesn't match the original collection's default collation. -changeStream = - caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "ABC"}}], - {resumeAfter: resumeToken, collation: {locale: "simple"}}); +changeStream = caseInsensitiveCollection.watch( + [{$match: {$or: [{"_id": resumeToken}, {"fullDocument.text": "ABC"}]}}], + {resumeAfter: resumeToken, collation: {locale: "simple"}}); assert.soon(() => changeStream.hasNext()); assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); @@ -304,9 +304,9 @@ if (!isChangeStreamPassthrough()) { // Test that a pipeline with an explicit collation is allowed to resume from before the // collection is dropped and recreated. -changeStream = - caseInsensitiveCollection.watch([{$match: {"fullDocument.text": "ABC"}}], - {resumeAfter: resumeToken, collation: {locale: "fr"}}); +changeStream = caseInsensitiveCollection.watch( + [{$match: {$or: [{"_id": resumeToken}, {"fullDocument.text": "ABC"}]}}], + {resumeAfter: resumeToken, collation: {locale: "fr"}}); assert.soon(() => changeStream.hasNext()); assert.docEq(changeStream.next().documentKey, {_id: "dropped_coll"}); diff --git a/jstests/change_streams/shell_helper.js b/jstests/change_streams/shell_helper.js index edf55161c6a..de7cf737280 100644 --- a/jstests/change_streams/shell_helper.js +++ b/jstests/change_streams/shell_helper.js @@ -131,8 +131,9 @@ if (!isMongos) { } // Only watch the "update" changes of the specific doc since the beginning. - changeStreamCursor = coll.watch([{$match: {documentKey: {_id: 1}, operationType: "update"}}], - {resumeAfter: resumeToken, batchSize: 2}); + changeStreamCursor = coll.watch( + [{$match: {$or: [{_id: resumeToken}, {documentKey: {_id: 1}, operationType: "update"}]}}], + {resumeAfter: resumeToken, batchSize: 2}); // Check the first batch. assert.eq(changeStreamCursor.objsLeftInBatch(), 2); diff --git a/jstests/noPassthrough/change_stream_pre_image_lookup_whole_db_whole_cluster.js b/jstests/noPassthrough/change_stream_pre_image_lookup_whole_db_whole_cluster.js index 8d0fc941020..078b85261a7 100644 --- a/jstests/noPassthrough/change_stream_pre_image_lookup_whole_db_whole_cluster.js +++ b/jstests/noPassthrough/change_stream_pre_image_lookup_whole_db_whole_cluster.js @@ -80,11 +80,12 @@ assert.throwsWithCode(function() { // don't trip the validation checks for the existence of the pre-image. for (let runOnDB of [testDB, adminDB]) { // Open a whole-db or whole-cluster stream that filters for the 'collWithPreImages' namespace. - const csCursor = runOnDB.watch([{$match: {"ns.coll": collWithPreImages.getName()}}], { - fullDocumentBeforeChange: "required", - resumeAfter: resumeToken, - allChangesForCluster: (runOnDB === adminDB) - }); + const csCursor = runOnDB.watch( + [{$match: {$or: [{_id: resumeToken}, {"ns.coll": collWithPreImages.getName()}]}}], { + fullDocumentBeforeChange: "required", + resumeAfter: resumeToken, + allChangesForCluster: (runOnDB === adminDB) + }); // The list of events and pre-images that we expect to see in the stream. const expectedPreImageEvents = [ diff --git a/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp b/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp index 0fff6070ba3..dbbf223f0ba 100644 --- a/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp +++ b/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp @@ -138,7 +138,7 @@ Value DocumentSourceChangeStreamTransform::serializeLegacy( _changeStreamSpec.getResumeAfter() || _changeStreamSpec.getStartAtOperationTime() || _changeStreamSpec.getStartAfter()); - return Value(Document{{getSourceName(), _changeStreamSpec.toBSON()}}); + return Value(Document{{DocumentSourceChangeStream::kStageName, _changeStreamSpec.toBSON()}}); } Value DocumentSourceChangeStreamCheckInvalidate::serializeLegacy( diff --git a/src/mongo/db/pipeline/document_source.cpp b/src/mongo/db/pipeline/document_source.cpp index 997f8c1f06a..3c183d9e847 100644 --- a/src/mongo/db/pipeline/document_source.cpp +++ b/src/mongo/db/pipeline/document_source.cpp @@ -34,9 +34,12 @@ #include "mongo/db/commands/feature_compatibility_version_documentation.h" #include "mongo/db/exec/document_value/value.h" #include "mongo/db/matcher/expression_algo.h" +#include "mongo/db/pipeline/document_source_add_fields.h" #include "mongo/db/pipeline/document_source_group.h" #include "mongo/db/pipeline/document_source_internal_shard_filter.h" #include "mongo/db/pipeline/document_source_match.h" +#include "mongo/db/pipeline/document_source_project.h" +#include "mongo/db/pipeline/document_source_replace_root.h" #include "mongo/db/pipeline/document_source_sample.h" #include "mongo/db/pipeline/document_source_sequential_document_cache.h" #include "mongo/db/pipeline/expression_context.h" @@ -239,13 +242,25 @@ bool DocumentSource::pushSampleBefore(Pipeline::SourceContainer::iterator itr, return false; } +bool DocumentSource::pushSingleDocumentTransformBefore(Pipeline::SourceContainer::iterator itr, + Pipeline::SourceContainer* container) { + auto singleDocTransform = + dynamic_cast<DocumentSourceSingleDocumentTransformation*>((*std::next(itr)).get()); + + if (constraints().canSwapWithSingleDocTransform && singleDocTransform) { + container->insert(itr, std::move(singleDocTransform)); + container->erase(std::next(itr)); + return true; + } + return false; +} + Pipeline::SourceContainer::iterator DocumentSource::optimizeAt( Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container) { invariant(*itr == this); - // Attempt to swap 'itr' with a subsequent $match or subsequent $sample. - if (std::next(itr) != container->end() && - (pushMatchBefore(itr, container) || pushSampleBefore(itr, container))) { + // Attempt to swap 'itr' with a subsequent stage, if applicable. + if (attemptToPushStageBefore(itr, container)) { // The stage before the pushed before stage may be able to optimize further, if there is // such a stage. return std::prev(itr) == container->begin() ? std::prev(itr) : std::prev(std::prev(itr)); diff --git a/src/mongo/db/pipeline/document_source.h b/src/mongo/db/pipeline/document_source.h index 9bc9f1cab2c..49cc5cddcfd 100644 --- a/src/mongo/db/pipeline/document_source.h +++ b/src/mongo/db/pipeline/document_source.h @@ -442,6 +442,31 @@ private: bool pushSampleBefore(Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container); + /** + * Attempts to push any kind of 'DocumentSourceSingleDocumentTransformation' stage directly + * ahead of the stage present at the 'itr' position if matches the constraints. Returns true if + * optimization was performed, false otherwise. + * + * Note that this optimization is oblivious to the transform function. The only stages that are + * eligible to swap are those that can safely swap with any transform. + */ + bool pushSingleDocumentTransformBefore(Pipeline::SourceContainer::iterator itr, + Pipeline::SourceContainer* container); + + /** + * Wraps various optimization methods and returns the call immediately if any one of them + * returns true. + */ + bool attemptToPushStageBefore(Pipeline::SourceContainer::iterator itr, + Pipeline::SourceContainer* container) { + if (std::next(itr) == container->end()) { + return false; + } + + return pushMatchBefore(itr, container) || pushSampleBefore(itr, container) || + pushSingleDocumentTransformBefore(itr, container); + } + public: /** * The non-virtual public interface for optimization. Attempts to do some generic optimizations diff --git a/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.cpp b/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.cpp index 7ea37e67da8..4ba82351811 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.cpp +++ b/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.cpp @@ -31,6 +31,8 @@ #include "mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h" +#include "mongo/db/query/query_feature_flags_gen.h" + namespace mongo { DocumentSourceChangeStreamEnsureResumeTokenPresent:: @@ -53,6 +55,34 @@ const char* DocumentSourceChangeStreamEnsureResumeTokenPresent::getSourceName() return kStageName.rawData(); } + +StageConstraints DocumentSourceChangeStreamEnsureResumeTokenPresent::constraints( + Pipeline::SplitState) const { + StageConstraints constraints{StreamType::kStreaming, + PositionRequirement::kNone, + // If this is parsed on mongos it should stay on mongos. If we're + // not in a sharded cluster then it's okay to run on mongod. + HostTypeRequirement::kLocalOnly, + DiskUseRequirement::kNoDiskUse, + FacetRequirement::kNotAllowed, + TransactionRequirement::kNotAllowed, + LookupRequirement::kNotAllowed, + UnionRequirement::kNotAllowed, + ChangeStreamRequirement::kChangeStreamStage}; + + // The '$match' and 'DocumentSourceSingleDocumentTransformation' stages can swap with this + // stage, allowing filtering and reshaping to occur earlier in the pipeline. For sharded cluster + // pipelines, swaps can allow $match and 'DocumentSourceSingleDocumentTransformation' stages to + // execute on the shards, providing inter-node parallelism and potentially reducing the amount + // of data sent form each shard to the mongoS. + if (feature_flags::gFeatureFlagChangeStreamsOptimization.isEnabledAndIgnoreFCV()) { + constraints.canSwapWithMatch = true; + constraints.canSwapWithSingleDocTransform = true; + } + + return constraints; +} + DocumentSource::GetNextResult DocumentSourceChangeStreamEnsureResumeTokenPresent::doGetNext() { // If we have already verified the resume token is present, return the next doc immediately. if (_resumeStatus == ResumeStatus::kSurpassedToken) { diff --git a/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h b/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h index 57b91107913..ae484dbff51 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h +++ b/src/mongo/db/pipeline/document_source_change_stream_ensure_resume_token_present.h @@ -39,22 +39,15 @@ namespace mongo { class DocumentSourceChangeStreamEnsureResumeTokenPresent final : public DocumentSourceChangeStreamCheckResumability { public: - static constexpr StringData kStageName = "$_internalEnsureResumeTokenPresent"_sd; + static constexpr StringData kStageName = "$_internalChangeStreamEnsureResumeTokenPresent"_sd; const char* getSourceName() const final; - StageConstraints constraints(Pipeline::SplitState) const final { - return {StreamType::kStreaming, - PositionRequirement::kNone, - // If this is parsed on mongos it should stay on mongos. If we're not in a sharded - // cluster then it's okay to run on mongod. - HostTypeRequirement::kLocalOnly, - DiskUseRequirement::kNoDiskUse, - FacetRequirement::kNotAllowed, - TransactionRequirement::kNotAllowed, - LookupRequirement::kNotAllowed, - UnionRequirement::kNotAllowed, - ChangeStreamRequirement::kChangeStreamStage}; + StageConstraints constraints(Pipeline::SplitState) const final; + + GetModPathsReturn getModifiedPaths() const final { + // This stage neither modifies nor renames any field. + return {GetModPathsReturn::Type::kFiniteSet, std::set<std::string>{}, {}}; } static boost::intrusive_ptr<DocumentSourceChangeStreamEnsureResumeTokenPresent> create( diff --git a/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp b/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp index 962e48fe4b8..d4c1d917799 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp +++ b/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.cpp @@ -55,8 +55,11 @@ bool isShardConfigEvent(const Document& eventDoc) { // with 4.2, even though we no longer rely on them to detect new shards. We swallow the event // here. We may wish to remove this mechanism entirely in 4.7+, or retain it for future cases // where a change stream is targeted to a subset of shards. See SERVER-44039 for details. - if (eventDoc[DocumentSourceChangeStream::kOperationTypeField].getStringData() == - DocumentSourceChangeStream::kNewShardDetectedOpType) { + + auto opType = eventDoc[DocumentSourceChangeStream::kOperationTypeField]; + + if (!opType.missing() && + opType.getStringData() == DocumentSourceChangeStream::kNewShardDetectedOpType) { // If the failpoint is enabled, throw the 'ChangeStreamToplogyChange' exception to the // client. This is used in testing to confirm that the swallowed 'kNewShardDetected' event // has reached the mongoS. @@ -87,6 +90,29 @@ DocumentSourceChangeStreamHandleTopologyChange::DocumentSourceChangeStreamHandle const boost::intrusive_ptr<ExpressionContext>& expCtx) : DocumentSource(kStageName, expCtx) {} +StageConstraints DocumentSourceChangeStreamHandleTopologyChange::constraints( + Pipeline::SplitState) const { + StageConstraints constraints{StreamType::kStreaming, + PositionRequirement::kNone, + HostTypeRequirement::kMongoS, + DiskUseRequirement::kNoDiskUse, + FacetRequirement::kNotAllowed, + TransactionRequirement::kNotAllowed, + LookupRequirement::kNotAllowed, + UnionRequirement::kNotAllowed, + ChangeStreamRequirement::kChangeStreamStage}; + + // Can be swapped with the '$match' and 'DocumentSourceSingleDocumentTransformation' stages and + // ensures that they get pushed down to the shards, as this stage bisects the change streams + // pipeline. + if (feature_flags::gFeatureFlagChangeStreamsOptimization.isEnabledAndIgnoreFCV()) { + constraints.canSwapWithMatch = true; + constraints.canSwapWithSingleDocTransform = true; + } + + return constraints; +} + DocumentSource::GetNextResult DocumentSourceChangeStreamHandleTopologyChange::doGetNext() { // For the first call to the 'doGetNext', the '_mergeCursors' will be null and must be // populated. We also resolve the original aggregation command from the expression context. diff --git a/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.h b/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.h index 1d240be5e3c..339d0db2f4d 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.h +++ b/src/mongo/db/pipeline/document_source_change_stream_handle_topology_change.h @@ -56,20 +56,19 @@ public: static boost::intrusive_ptr<DocumentSourceChangeStreamHandleTopologyChange> create( const boost::intrusive_ptr<ExpressionContext>&); + const char* getSourceName() const final { + return kStageName.rawData(); + } + Value serialize(boost::optional<ExplainOptions::Verbosity> explain) const final { return (explain ? Value(Document{{kStageName, Value()}}) : Value()); } - virtual StageConstraints constraints(Pipeline::SplitState) const { - return {StreamType::kStreaming, - PositionRequirement::kNone, - HostTypeRequirement::kMongoS, - DiskUseRequirement::kNoDiskUse, - FacetRequirement::kNotAllowed, - TransactionRequirement::kNotAllowed, - LookupRequirement::kNotAllowed, - UnionRequirement::kNotAllowed, - ChangeStreamRequirement::kChangeStreamStage}; + StageConstraints constraints(Pipeline::SplitState) const final; + + GetModPathsReturn getModifiedPaths() const final { + // This stage neither modifies nor renames any field. + return {GetModPathsReturn::Type::kFiniteSet, std::set<std::string>{}, {}}; } boost::optional<DistributedPlanLogic> distributedPlanLogic() final { diff --git a/src/mongo/db/pipeline/document_source_change_stream_oplog_match.cpp b/src/mongo/db/pipeline/document_source_change_stream_oplog_match.cpp index 5dd28772c6c..a37869bd035 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_oplog_match.cpp +++ b/src/mongo/db/pipeline/document_source_change_stream_oplog_match.cpp @@ -212,7 +212,7 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceChangeStreamOplogMatch::creat const char* DocumentSourceChangeStreamOplogMatch::getSourceName() const { // This is used in error reporting, particularly if we find this stage in a position other // than first, so report the name as $changeStream. - return DocumentSourceChangeStream::kStageName.rawData(); + return kStageName.rawData(); } StageConstraints DocumentSourceChangeStreamOplogMatch::constraints( diff --git a/src/mongo/db/pipeline/document_source_change_stream_test.cpp b/src/mongo/db/pipeline/document_source_change_stream_test.cpp index a340096f5e7..7452d9f7d98 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp +++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp @@ -267,7 +267,7 @@ public: // Check the oplog entry is transformed correctly. auto transform = stages[2].get(); ASSERT(transform); - ASSERT_EQ(string(transform->getSourceName()), DSChangeStream::kStageName); + ASSERT(dynamic_cast<DocumentSourceChangeStreamTransform*>(transform)); // Create mock stage and insert at the front of the stages. auto mock = DocumentSourceMock::createForTest(D(entry), getExpCtx()); @@ -466,6 +466,44 @@ public: } }; +class ChangeStreamPipelineOptimizationTest : public ChangeStreamStageTest { +public: + explicit ChangeStreamPipelineOptimizationTest() : ChangeStreamStageTest() {} + + void run() { + RAIIServerParameterControllerForTest controller("featureFlagChangeStreamsOptimization", + true); + ASSERT(getCSOptimizationFeatureFlagValue()); + ChangeStreamStageTest::run(); + } + + std::unique_ptr<Pipeline, PipelineDeleter> buildTestPipeline( + const std::vector<BSONObj>& rawPipeline) { + auto expCtx = getExpCtx(); + expCtx->ns = NamespaceString("a.collection"); + expCtx->inMongos = true; + + auto pipeline = Pipeline::parse(rawPipeline, expCtx); + pipeline->optimizePipeline(); + + return pipeline; + } + + void assertStagesNameOrder(std::unique_ptr<Pipeline, PipelineDeleter> pipeline, + const std::vector<std::string> expectedStages) { + ASSERT_EQ(pipeline->getSources().size(), expectedStages.size()); + + auto stagesItr = pipeline->getSources().begin(); + auto expectedStagesItr = expectedStages.begin(); + + while (expectedStagesItr != expectedStages.end()) { + ASSERT_EQ(*expectedStagesItr, stagesItr->get()->getSourceName()); + ++expectedStagesItr; + ++stagesItr; + } + } +}; + TEST_F(ChangeStreamStageTest, ShouldRejectNonObjectArg) { auto expCtx = getExpCtx(); @@ -2166,7 +2204,6 @@ template <typename Stage, typename StageSpec> void validateDocumentSourceStageSerialization( StageSpec spec, BSONObj specAsBSON, const boost::intrusive_ptr<ExpressionContext>& expCtx) { auto stage = Stage::createFromBson(specAsBSON.firstElement(), expCtx); - vector<Value> serialization; stage->serializeToArray(serialization); if (getCSOptimizationFeatureFlagValue()) { @@ -3304,5 +3341,513 @@ TEST_F(ChangeStreamStageDBTest, StartAfterSucceedsEvenIfResumeTokenDoesNotContai BSON("$changeStream" << BSON("startAfter" << resumeToken))); } +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSingleMatch) { + // + // Tests that the single '$match' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = { + fromjson("{$changeStream: {}}"), + fromjson("{$match: {operationType: 'insert'}}"), + }; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$match", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithMultipleMatch) { + // + // Tests that multiple '$match' gets merged and promoted before the + // '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$match: {operationType: 'insert'}}"), + fromjson("{$match: {operationType: 'delete'}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$match", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithMultipleMatchAndResumeToken) { + // + // Tests that multiple '$match' gets merged and promoted before the + // '$_internalUpdateOnAddShard' if resume token if present. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = { + BSON("$changeStream" << BSON("resumeAfter" << resumeToken)), + BSON("$match" << BSON("operationType" + << "insert")), + BSON("$match" << BSON("operationType" + << "insert"))}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$match", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSingleProject) { + // + // Tests that the single'$project' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$project: {operationType: 1}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$project", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithMultipleProject) { + // + // Tests that multiple '$project' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$project: {operationType: 1}}"), + fromjson("{$project: {fullDocument: 1}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$project", + "$project", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithMultipleProjectAndResumeToken) { + // + // Tests that multiple '$project' gets promoted before the '$_internalUpdateOnAddShard' if + // resume token is present. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = { + BSON("$changeStream" << BSON("resumeAfter" << resumeToken)), + BSON("$project" << BSON("operationType" << 1)), + BSON("$project" << BSON("fullDocument" << 1))}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$project", + "$project", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithProjectMatchAndResumeToken) { + // + // Tests that a '$project' followed by a '$match' gets optimized and they get promoted before + // the '$_internalUpdateOnAddShard'. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = { + BSON("$changeStream" << BSON("resumeAfter" << resumeToken)), + BSON("$project" << BSON("operationType" << 1)), + BSON("$match" << BSON("operationType" + << "insert"))}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$match", + "$project", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSingleUnset) { + // + // Tests that the single'$unset' gets promoted before the '$_internalUpdateOnAddShard' as + // '$project'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$unset: 'operationType'}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$project", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithMultipleUnset) { + // + // Tests that multiple '$unset' gets promoted before the '$_internalUpdateOnAddShard' as + // '$project'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$unset: 'operationType'}"), + fromjson("{$unset: 'fullDocument'}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$project", + "$project", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithUnsetAndResumeToken) { + // + // Tests that the '$unset' gets promoted before the '$_internalUpdateOnAddShard' as '$project' + // even if resume token is present. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = {BSON("$changeStream" + << BSON("resumeAfter" << resumeToken)), + BSON("$unset" + << "operationType")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$project", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSingleAddFields) { + // + // Tests that the single'$addFields' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$addFields: {stockPrice: 100}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$addFields", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithMultipleAddFields) { + // + // Tests that multiple '$addFields' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$addFields: {stockPrice: 100}}"), + fromjson("{$addFields: {quarter: 'Q1'}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$addFields", + "$addFields", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithAddFieldsAndResumeToken) { + // + // Tests that the '$addFields' gets promoted before the '$_internalUpdateOnAddShard' if + // resume token is present. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = { + BSON("$changeStream" << BSON("resumeAfter" << resumeToken)), + BSON("$addFields" << BSON("stockPrice" << 100))}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$addFields", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSingleSet) { + // + // Tests that the single'$set' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$set: {stockPrice: 100}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$set", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithMultipleSet) { + // + // Tests that multiple '$set' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$set: {stockPrice: 100}}"), + fromjson("{$set: {quarter: 'Q1'}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$set", + "$set", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSetAndResumeToken) { + // + // Tests that the '$set' gets promoted before the '$_internalUpdateOnAddShard' if + // resume token is present. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = { + BSON("$changeStream" << BSON("resumeAfter" << resumeToken)), + BSON("$set" << BSON("stockPrice" << 100))}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$set", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSingleReplaceRoot) { + // + // Tests that the single'$replaceRoot' gets promoted before the '$_internalUpdateOnAddShard'. + // + const std::vector<BSONObj> rawPipeline = { + fromjson("{$changeStream: {}}"), fromjson("{$replaceRoot: {newRoot: '$fullDocument'}}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$replaceRoot", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithReplaceRootAndResumeToken) { + // + // Tests that the '$replaceRoot' gets promoted before the '$_internalUpdateOnAddShard' if + // resume token is present. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = { + BSON("$changeStream" << BSON("resumeAfter" << resumeToken)), + BSON("$replaceRoot" << BSON("newRoot" + << "$fullDocument"))}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$replaceRoot", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithSingleReplaceWith) { + // + // Tests that the single '$replaceWith' gets promoted before the '$_internalUpdateOnAddShard' as + // '$replaceRoot'. + // + const std::vector<BSONObj> rawPipeline = {fromjson("{$changeStream: {}}"), + fromjson("{$replaceWith: '$fullDocument'}")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$replaceRoot", + "$_internalChangeStreamHandleTopologyChange"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithReplaceWithAndResumeToken) { + // + // Tests that the '$replaceWith' gets promoted before the '$_internalUpdateOnAddShard' if + // resume token is present as '$replaceRoot'. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = {BSON("$changeStream" + << BSON("resumeAfter" << resumeToken)), + BSON("$replaceWith" + << "$fullDocument")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$replaceRoot", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + +TEST_F(ChangeStreamPipelineOptimizationTest, ChangeStreamWithAllStagesAndResumeToken) { + // + // Tests that when all allowed stages are included along with the resume token, the final + // pipeline gets optimized. + // + auto resumeToken = makeResumeToken(kDefaultTs, testUuid()); + + const std::vector<BSONObj> rawPipeline = {BSON("$changeStream" + << BSON("resumeAfter" << resumeToken)), + BSON("$project" << BSON("operationType" << 1)), + BSON("$unset" + << "_id"), + BSON("$addFields" << BSON("stockPrice" << 100)), + BSON("$set" + << BSON("fullDocument.stockPrice" << 100)), + BSON("$match" << BSON("operationType" + << "insert")), + BSON("$replaceRoot" << BSON("newRoot" + << "$fullDocument")), + BSON("$replaceWith" + << "fullDocument.stockPrice")}; + + auto pipeline = buildTestPipeline(rawPipeline); + + assertStagesNameOrder(std::move(pipeline), + {"$_internalChangeStreamOplogMatch", + "$_internalChangeStreamUnwindTransaction", + "$_internalChangeStreamTransform", + "$_internalChangeStreamCheckInvalidate", + "$_internalChangeStreamCheckResumability", + "$_internalChangeStreamCheckTopologyChange", + "$match", + "$project", + "$project", + "$addFields", + "$set", + "$replaceRoot", + "$replaceRoot", + "$_internalChangeStreamHandleTopologyChange", + "$_internalChangeStreamEnsureResumeTokenPresent"}); +} + } // namespace } // namespace mongo diff --git a/src/mongo/db/pipeline/document_source_change_stream_transform.cpp b/src/mongo/db/pipeline/document_source_change_stream_transform.cpp index 229b9229565..c8eacdda7df 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_transform.cpp +++ b/src/mongo/db/pipeline/document_source_change_stream_transform.cpp @@ -390,7 +390,7 @@ Document DocumentSourceChangeStreamTransform::applyTransformation(const Document Value DocumentSourceChangeStreamTransform::serializeLatest( boost::optional<ExplainOptions::Verbosity> explain) const { if (explain) { - return Value(Document{{getSourceName(), + return Value(Document{{DocumentSourceChangeStream::kStageName, Document{{"stage"_sd, "internalTransform"_sd}, {"options"_sd, _changeStreamSpec.toBSON()}}}}); } diff --git a/src/mongo/db/pipeline/document_source_change_stream_transform.h b/src/mongo/db/pipeline/document_source_change_stream_transform.h index 46fe08b7a49..f4782ca89cf 100644 --- a/src/mongo/db/pipeline/document_source_change_stream_transform.h +++ b/src/mongo/db/pipeline/document_source_change_stream_transform.h @@ -64,8 +64,8 @@ public: return boost::none; } - const char* getSourceName() const { - return DocumentSourceChangeStream::kStageName.rawData(); + const char* getSourceName() const final { + return kStageName.rawData(); } protected: diff --git a/src/mongo/db/pipeline/document_source_limit_test.cpp b/src/mongo/db/pipeline/document_source_limit_test.cpp index 04aa426bd1e..899dbe4de42 100644 --- a/src/mongo/db/pipeline/document_source_limit_test.cpp +++ b/src/mongo/db/pipeline/document_source_limit_test.cpp @@ -37,6 +37,7 @@ #include "mongo/db/pipeline/document_source_limit.h" #include "mongo/db/pipeline/document_source_match.h" #include "mongo/db/pipeline/document_source_mock.h" +#include "mongo/db/pipeline/document_source_project.h" #include "mongo/db/pipeline/pipeline.h" #include "mongo/unittest/unittest.h" @@ -90,6 +91,23 @@ TEST_F(DocumentSourceLimitTest, TwoLimitStagesShouldCombineIntoOne) { ASSERT_EQUALS(1U, container.size()); } +TEST_F(DocumentSourceLimitTest, DoesNotPushProjectBeforeSelf) { + Pipeline::SourceContainer container; + auto limit = DocumentSourceLimit::create(getExpCtx(), 10); + auto project = + DocumentSourceProject::create(BSON("fullDocument" << true), getExpCtx(), "$project"_sd); + + container.push_back(limit); + container.push_back(project); + + limit->optimizeAt(container.begin(), &container); + + ASSERT_EQUALS(2U, container.size()); + ASSERT(dynamic_cast<DocumentSourceLimit*>(container.begin()->get())); + ASSERT(dynamic_cast<DocumentSourceSingleDocumentTransformation*>( + std::next(container.begin())->get())); +} + TEST_F(DocumentSourceLimitTest, DisposeShouldCascadeAllTheWayToSource) { auto source = DocumentSourceMock::createForTest({"{a: 1}", "{a: 1}"}, getExpCtx()); diff --git a/src/mongo/db/pipeline/document_source_match_test.cpp b/src/mongo/db/pipeline/document_source_match_test.cpp index 703512557d6..6ab275e0417 100644 --- a/src/mongo/db/pipeline/document_source_match_test.cpp +++ b/src/mongo/db/pipeline/document_source_match_test.cpp @@ -41,6 +41,7 @@ #include "mongo/db/pipeline/aggregation_context_fixture.h" #include "mongo/db/pipeline/document_source_match.h" #include "mongo/db/pipeline/document_source_mock.h" +#include "mongo/db/pipeline/document_source_project.h" #include "mongo/db/pipeline/pipeline.h" #include "mongo/logv2/log.h" #include "mongo/unittest/death_test.h" @@ -492,6 +493,23 @@ TEST_F(DocumentSourceMatchTest, MultipleMatchStagesShouldCombineIntoOne) { "{c:1}]}")); } +TEST_F(DocumentSourceMatchTest, DoesNotPushProjectBeforeSelf) { + Pipeline::SourceContainer container; + auto match = DocumentSourceMatch::create(BSON("_id" << 1), getExpCtx()); + auto project = + DocumentSourceProject::create(BSON("fullDocument" << true), getExpCtx(), "$project"_sd); + + container.push_back(match); + container.push_back(project); + + match->optimizeAt(container.begin(), &container); + + ASSERT_EQUALS(2U, container.size()); + ASSERT(dynamic_cast<DocumentSourceMatch*>(container.begin()->get())); + ASSERT(dynamic_cast<DocumentSourceSingleDocumentTransformation*>( + std::next(container.begin())->get())); +} + TEST_F(DocumentSourceMatchTest, ShouldPropagatePauses) { auto match = DocumentSourceMatch::create(BSON("a" << 1), getExpCtx()); auto mock = diff --git a/src/mongo/db/pipeline/document_source_sort_test.cpp b/src/mongo/db/pipeline/document_source_sort_test.cpp index 980fc537f97..9227cc35f71 100644 --- a/src/mongo/db/pipeline/document_source_sort_test.cpp +++ b/src/mongo/db/pipeline/document_source_sort_test.cpp @@ -43,6 +43,7 @@ #include "mongo/db/pipeline/aggregation_context_fixture.h" #include "mongo/db/pipeline/dependencies.h" #include "mongo/db/pipeline/document_source_mock.h" +#include "mongo/db/pipeline/document_source_project.h" #include "mongo/db/pipeline/document_source_sort.h" #include "mongo/db/pipeline/pipeline.h" #include "mongo/unittest/temp_dir.h" @@ -163,6 +164,23 @@ TEST_F(DocumentSourceSortTest, SortWithLimit) { ASSERT(dynamic_cast<DocumentSourceLimit*>(sort()->distributedPlanLogic()->mergingStage.get())); } +TEST_F(DocumentSourceSortTest, DoesNotPushProjectBeforeSelf) { + Pipeline::SourceContainer container; + createSort(BSON("_id" << 1)); + auto project = + DocumentSourceProject::create(BSON("fullDocument" << true), getExpCtx(), "$project"_sd); + + container.push_back(sort()); + container.push_back(project); + + sort()->optimizeAt(container.begin(), &container); + + ASSERT_EQUALS(2U, container.size()); + ASSERT(dynamic_cast<DocumentSourceSort*>(container.begin()->get())); + ASSERT(dynamic_cast<DocumentSourceSingleDocumentTransformation*>( + std::next(container.begin())->get())); +} + TEST_F(DocumentSourceSortTest, Dependencies) { createSort(BSON("a" << 1 << "b.c" << -1)); DepsTracker dependencies; diff --git a/src/mongo/db/pipeline/pipeline_test.cpp b/src/mongo/db/pipeline/pipeline_test.cpp index 37586a16abd..fb81a3ed888 100644 --- a/src/mongo/db/pipeline/pipeline_test.cpp +++ b/src/mongo/db/pipeline/pipeline_test.cpp @@ -76,7 +76,7 @@ using std::vector; const NamespaceString kTestNss = NamespaceString("a.collection"); size_t getChangeStreamStageSize() { - return (feature_flags::gFeatureFlagChangeStreamsOptimization.isEnabledAndIgnoreFCV() ? 5 : 6); + return (feature_flags::gFeatureFlagChangeStreamsOptimization.isEnabledAndIgnoreFCV() ? 6 : 6); } void setMockReplicationCoordinatorOnOpCtx(OperationContext* opCtx) { diff --git a/src/mongo/db/pipeline/plan_executor_pipeline.cpp b/src/mongo/db/pipeline/plan_executor_pipeline.cpp index e491d99afdc..3f167f28530 100644 --- a/src/mongo/db/pipeline/plan_executor_pipeline.cpp +++ b/src/mongo/db/pipeline/plan_executor_pipeline.cpp @@ -176,11 +176,6 @@ void PlanExecutorPipeline::_performChangeStreamsAccounting(const boost::optional } void PlanExecutorPipeline::_validateChangeStreamsResumeToken(const Document& event) const { - // If we are producing output to be merged on mongoS, then no stages can have modified the _id. - if (_expCtx->needsMerge) { - return; - } - // Confirm that the document _id field matches the original resume token in the sort key field. auto eventBSON = event.toBson(); auto resumeToken = event.metadata().getSortKey(); diff --git a/src/mongo/db/pipeline/stage_constraints.h b/src/mongo/db/pipeline/stage_constraints.h index 661b9954195..b520dfdbbdb 100644 --- a/src/mongo/db/pipeline/stage_constraints.h +++ b/src/mongo/db/pipeline/stage_constraints.h @@ -328,6 +328,10 @@ struct StageConstraints { // documents because our implementation of $sample shuffles the order bool canSwapWithSkippingOrLimitingStage = false; + // If true, then any stage of kind 'DocumentSourceSingleDocumentTransformation' can be swapped + // ahead of this stage. + bool canSwapWithSingleDocTransform = false; + // Indicates that a stage is allowed within a pipeline-stlye update. bool isAllowedWithinUpdatePipeline = false; diff --git a/src/mongo/s/query/async_results_merger.cpp b/src/mongo/s/query/async_results_merger.cpp index d6eca0d9b1b..236729e2359 100644 --- a/src/mongo/s/query/async_results_merger.cpp +++ b/src/mongo/s/query/async_results_merger.cpp @@ -489,6 +489,11 @@ Status AsyncResultsMerger::_askForNextBatch(WithLock, size_t remoteIndex) { Status AsyncResultsMerger::scheduleGetMores() { stdx::lock_guard<Latch> lk(_mutex); + + if (feature_flags::gFeatureFlagChangeStreamsOptimization.isEnabledAndIgnoreFCV()) { + _assertNotInvalidated(lk); + } + return _scheduleGetMores(lk); } |