SERVER-20214 Stop supporting majority read concern level with MapReduce and Aggregation $out

6 files changed, 54 insertions, 11 deletions

diff --git a/jstests/noPassthrough/read_majority.js b/jstests/noPassthrough/read_majority.js
index be3572fab13..93b92043b05 100644
--- a/jstests/noPassthrough/read_majority.js
+++ b/jstests/noPassthrough/read_majority.js
@@ -22,6 +22,12 @@ function getReadMajorityCursor() {
     return new DBCommandCursor(db.getMongo(), res, 2);
 }
 
+function getReadMajorityAggCursor() {
+    var res = t.runCommand('aggregate', {cursor:{batchSize: 2}, readConcern: {level: "majority"}});
+    assert.commandWorked(res);
+    return new DBCommandCursor(db.getMongo(), res, 2);
+}
+
 function getReadMajorityExplainPlan(query) {
     var res = db.runCommand({explain: {find: t.getName(), filter: query},
                              readConcern: {level: "majority"}});
@@ -71,7 +77,6 @@ assertNoReadMajoritySnapshotAvailable();
 assert.writeOK(t.update({}, {$set: {version: 4}}, false, true));
 var snapshot4 = assert.commandWorked(db.adminCommand("makeSnapshot")).name;
 
-
 // Collection didn't exist in snapshot 1.
 assert.commandWorked(db.adminCommand({"setCommittedSnapshot": snapshot1}));
 assertNoReadMajoritySnapshotAvailable();
@@ -79,9 +84,17 @@ assertNoReadMajoritySnapshotAvailable();
 // Collection existed but was empty in snapshot 2.
 assert.commandWorked(db.adminCommand({"setCommittedSnapshot": snapshot2}));
 assert.eq(getReadMajorityCursor().itcount(), 0);
+assert.eq(getReadMajorityAggCursor().itcount(), 0);
 
 // In snapshot 3 the collection was filled with {version: 3} documents.
 assert.commandWorked(db.adminCommand({"setCommittedSnapshot": snapshot3}));
+assert.eq(getReadMajorityAggCursor().itcount(), 10);
+getReadMajorityAggCursor().forEach(function(doc) {
+    // Note: agg uses internal batching so can't reliably test flipping snapshot. However, it uses
+    // the same mechanism as find, so if one works, both should.
+    assert.eq(doc.version, 3)
+});
+
 assert.eq(getReadMajorityCursor().itcount(), 10);
 var cursor = getReadMajorityCursor(); // Note: uses batchsize=2.
 assert.eq(cursor.next().version, 3);
@@ -100,6 +113,7 @@ assert.eq(cursor.next().version, 4);
 t.ensureIndex({version: 1});
 if (false) {
     // disabled until SERVER-20439 is implemented.
+    assert.eq(getReadMajorityAggCursor().itcount(), 10);
     assert.eq(getReadMajorityCursor().itcount(), 10);
     assert(!isIxscan(getReadMajorityExplainPlan({version: 1})));
 } else {
@@ -144,12 +158,14 @@ newSnapshot = assert.commandWorked(db.adminCommand("makeSnapshot")).name;
 assertNoReadMajoritySnapshotAvailable();
 assert.commandWorked(db.adminCommand({"setCommittedSnapshot": newSnapshot}));
 assert.eq(getReadMajorityCursor().itcount(), 10);
+assert.eq(getReadMajorityAggCursor().itcount(), 10);
 
 // Dropping the collection is visible in the committed snapshot, even though it hasn't been marked
 // committed yet. This is allowed by the current specification even though it violates strict
 // read-committed semantics since we don't guarantee them on metadata operations.
 t.drop();
 assert.eq(getReadMajorityCursor().itcount(), 0);
+assert.eq(getReadMajorityAggCursor().itcount(), 0);
 
 // Creating a new collection with the same name hides the collection until that operation is in the
 // committed view.
@@ -159,6 +175,20 @@ newSnapshot = assert.commandWorked(db.adminCommand("makeSnapshot")).name;
 assertNoReadMajoritySnapshotAvailable();
 assert.commandWorked(db.adminCommand({"setCommittedSnapshot": newSnapshot}));
 assert.eq(getReadMajorityCursor().itcount(), 1);
+assert.eq(getReadMajorityAggCursor().itcount(), 1);
+
+// Commands that only support read concern 'local', (such as ping) must work when it is explicitly
+// specified and fail when 'majority' is specified.
+assert.commandWorked(db.adminCommand({ping: 1, readConcern: {level: 'local'}}));
+var res = assert.commandFailed(db.adminCommand({ping: 1, readConcern: {level: 'majority'}}));
+assert.eq(res.code, ErrorCodes.InvalidOptions);
+
+// Agg $out also doesn't support read concern majority.
+assert.commandWorked(t.runCommand('aggregate', {pipeline: [{$out: 'out'}],
+                                                readConcern: {level: 'local'}}));
+var res = assert.commandFailed(t.runCommand('aggregate', {pipeline: [{$out: 'out'}],
+                                                          readConcern: {level: 'majority'}}));
+assert.eq(res.code, ErrorCodes.InvalidOptions);
 
 MongoRunner.stopMongod(testServer);
 }());
diff --git a/src/mongo/db/commands/mr.cpp b/src/mongo/db/commands/mr.cpp
index 772f758fc6c..1c304dc5c25 100644
--- a/src/mongo/db/commands/mr.cpp
+++ b/src/mongo/db/commands/mr.cpp
@@ -1247,10 +1247,6 @@ public:
         return true;
     }
 
-    bool supportsReadConcern() const final {
-        return true;
-    }
-
     virtual void help(stringstream& help) const {
         help << "Run a map/reduce operation on the server.\n";
         help << "Note this is used for aggregation, not querying, in MongoDB.\n";
diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index 58167b69f6e..f2089446752 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -143,6 +143,7 @@ env.Library(
         'document_value',
         '$BUILD_DIR/mongo/db/auth/authorization_manager_global',
         '$BUILD_DIR/mongo/db/auth/authcore',
+        '$BUILD_DIR/mongo/db/repl/read_concern_args',
     ]
 )
 
diff --git a/src/mongo/db/pipeline/pipeline.cpp b/src/mongo/db/pipeline/pipeline.cpp
index 413bf9bf07b..44763f86343 100644
--- a/src/mongo/db/pipeline/pipeline.cpp
+++ b/src/mongo/db/pipeline/pipeline.cpp
@@ -43,6 +43,7 @@
 #include "mongo/db/pipeline/document_source.h"
 #include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/expression_context.h"
+#include "mongo/db/repl/read_concern_args.h"
 #include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
@@ -68,6 +69,7 @@ intrusive_ptr<Pipeline> Pipeline::parseCommand(string& errmsg,
                                                const intrusive_ptr<ExpressionContext>& pCtx) {
     intrusive_ptr<Pipeline> pPipeline(new Pipeline(pCtx));
     vector<BSONElement> pipeline;
+    repl::ReadConcernArgs readConcernArgs;
 
     /* gather the specification for the aggregation */
     for (BSONObj::iterator cmdIterator = cmdObj.begin(); cmdIterator.more();) {
@@ -84,8 +86,8 @@ intrusive_ptr<Pipeline> Pipeline::parseCommand(string& errmsg,
             continue;
         }
 
-        // readConcern is also for the command processor.
-        if (str::equals(pFieldName, "readConcern")) {
+        if (pFieldName == repl::ReadConcernArgs::kReadConcernFieldName) {
+            uassertStatusOK(readConcernArgs.initialize(cmdElement));
             continue;
         }
 
@@ -160,6 +162,10 @@ intrusive_ptr<Pipeline> Pipeline::parseCommand(string& errmsg,
 
         if (dynamic_cast<DocumentSourceOut*>(sources.back().get())) {
             uassert(16991, "$out can only be the final stage in the pipeline", iStep == nSteps - 1);
+
+            uassert(ErrorCodes::InvalidOptions,
+                    "$out can only be used with the 'local' read concern level",
+                    readConcernArgs.getLevel() == repl::ReadConcernLevel::kLocalReadConcern);
         }
     }
 
diff --git a/src/mongo/db/repl/read_concern_args.cpp b/src/mongo/db/repl/read_concern_args.cpp
index 1389a7380a1..20b179f8033 100644
--- a/src/mongo/db/repl/read_concern_args.cpp
+++ b/src/mongo/db/repl/read_concern_args.cpp
@@ -67,13 +67,13 @@ OpTime ReadConcernArgs::getOpTime() const {
     return _opTime.value_or(OpTime());
 }
 
-Status ReadConcernArgs::initialize(const BSONObj& cmdObj) {
-    auto readConcernElem = cmdObj[ReadConcernArgs::kReadConcernFieldName];
-
+Status ReadConcernArgs::initialize(const BSONElement& readConcernElem) {
     if (readConcernElem.eoo()) {
         return Status::OK();
     }
 
+    dassert(readConcernElem.fieldNameStringData() == kReadConcernFieldName);
+
     if (!readConcernElem.isABSONObj()) {
         return Status(ErrorCodes::FailedToParse,
                       str::stream() << kReadConcernFieldName << " field should be an object");
diff --git a/src/mongo/db/repl/read_concern_args.h b/src/mongo/db/repl/read_concern_args.h
index a010d9c8793..2bb1a07c3a4 100644
--- a/src/mongo/db/repl/read_concern_args.h
+++ b/src/mongo/db/repl/read_concern_args.h
@@ -32,6 +32,7 @@
 #include <string>
 
 #include "mongo/base/status.h"
+#include "mongo/db/json.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/util/time_support.h"
 
@@ -63,7 +64,16 @@ public:
      *    }
      * }
      */
-    Status initialize(const BSONObj& cmdObj);
+    Status initialize(const BSONObj& cmdObj) {
+        return initialize(cmdObj[kReadConcernFieldName]);
+    }
+
+    /**
+     * Initializes the object from the readConcern element in a command object.
+     * Use this if you are already iterating over the fields in the command object.
+     * This method correctly handles missing BSONElements.
+     */
+    Status initialize(const BSONElement& readConcernElem);
 
     /**
      * Appends level and afterOpTime.