SERVER-35776 Basic 4.0-4.2 up/downgrade for change streams

1 files changed, 14 insertions, 5 deletions

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 c1432f4d960..a7f6e01d235 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_transform.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_transform.cpp
@@ -68,16 +68,21 @@ constexpr auto checkValueType = &DocumentSourceChangeStream::checkValueType;
 }  // namespace
 
 boost::intrusive_ptr<DocumentSourceChangeStreamTransform>
-DocumentSourceChangeStreamTransform::create(const boost::intrusive_ptr<ExpressionContext>& expCtx,
-                                            BSONObj changeStreamSpec) {
-    return new DocumentSourceChangeStreamTransform(expCtx, changeStreamSpec);
+DocumentSourceChangeStreamTransform::create(
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    const ServerGlobalParams::FeatureCompatibility::Version& fcv,
+    BSONObj changeStreamSpec) {
+    return new DocumentSourceChangeStreamTransform(expCtx, fcv, changeStreamSpec);
 }
 
 DocumentSourceChangeStreamTransform::DocumentSourceChangeStreamTransform(
-    const boost::intrusive_ptr<ExpressionContext>& expCtx, BSONObj changeStreamSpec)
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    const ServerGlobalParams::FeatureCompatibility::Version& fcv,
+    BSONObj changeStreamSpec)
     : DocumentSource(expCtx),
       _changeStreamSpec(changeStreamSpec.getOwned()),
-      _isIndependentOfAnyCollection(expCtx->ns.isCollectionlessAggregateNS()) {
+      _isIndependentOfAnyCollection(expCtx->ns.isCollectionlessAggregateNS()),
+      _fcv(fcv) {
 
     _nsRegex.emplace(DocumentSourceChangeStream::getNsRegexForChangeStream(expCtx->ns));
 
@@ -174,6 +179,10 @@ ResumeTokenData DocumentSourceChangeStreamTransform::getResumeToken(Value ts,
     if (!uuid.missing())
         resumeTokenData.uuid = uuid.getUuid();
 
+    if (_fcv < ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo42) {
+        resumeTokenData.version = 0;
+    }
+
     return resumeTokenData;
 }