SERVER-31665 Use correct read concern/preference during update lookup

17 files changed, 68 insertions, 12 deletions

diff --git a/src/mongo/db/pipeline/aggregation_request.h b/src/mongo/db/pipeline/aggregation_request.h
index b34fc383d20..d6b87a5f183 100644
--- a/src/mongo/db/pipeline/aggregation_request.h
+++ b/src/mongo/db/pipeline/aggregation_request.h
@@ -274,7 +274,7 @@ private:
     // {$hint: <String>}, where <String> is the index name hinted.
     BSONObj _hint;
 
-    // The comment parameter attached to this aggregation.
+    // The comment parameter attached to this aggregation, empty if not set.
     std::string _comment;
 
     BSONObj _readConcern;
diff --git a/src/mongo/db/pipeline/document_source.h b/src/mongo/db/pipeline/document_source.h
index 0ecd0fcf92a..4585c68637c 100644
--- a/src/mongo/db/pipeline/document_source.h
+++ b/src/mongo/db/pipeline/document_source.h
@@ -847,9 +847,11 @@ public:
          * no matching documents were found, including cases where the given namespace does not
          * exist.
          */
-        virtual boost::optional<Document> lookupSingleDocument(const NamespaceString& nss,
-                                                               UUID collectionUUID,
-                                                               const Document& documentKey) = 0;
+        virtual boost::optional<Document> lookupSingleDocument(
+            const NamespaceString& nss,
+            UUID collectionUUID,
+            const Document& documentKey,
+            boost::optional<BSONObj> readConcern) = 0;
 
         /**
          * Returns a vector of all local cursors.
diff --git a/src/mongo/db/pipeline/document_source_lookup_change_post_image.cpp b/src/mongo/db/pipeline/document_source_lookup_change_post_image.cpp
index 7752fbfd751..b616b5ec659 100644
--- a/src/mongo/db/pipeline/document_source_lookup_change_post_image.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup_change_post_image.cpp
@@ -105,9 +105,15 @@ Value DocumentSourceLookupChangePostImage::lookupPostImage(const Document& updat
     auto resumeToken =
         ResumeToken::parse(updateOp[DocumentSourceChangeStream::kIdField].getDocument());
 
+    const auto readConcern = pExpCtx->inMongos
+        ? boost::optional<BSONObj>(BSON("level"
+                                        << "majority"
+                                        << "afterClusterTime"
+                                        << resumeToken.getData().clusterTime))
+        : boost::none;
     invariant(resumeToken.getData().uuid);
-    auto lookedUpDoc =
-        _mongoProcessInterface->lookupSingleDocument(nss, *resumeToken.getData().uuid, documentKey);
+    auto lookedUpDoc = _mongoProcessInterface->lookupSingleDocument(
+        nss, *resumeToken.getData().uuid, documentKey, readConcern);
 
     // Check whether the lookup returned any documents. Even if the lookup itself succeeded, it may
     // not have returned any results if the document was deleted in the time since the update op.
diff --git a/src/mongo/db/pipeline/document_source_lookup_change_post_image_test.cpp b/src/mongo/db/pipeline/document_source_lookup_change_post_image_test.cpp
index 3d35f875619..1d003c9c09e 100644
--- a/src/mongo/db/pipeline/document_source_lookup_change_post_image_test.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup_change_post_image_test.cpp
@@ -115,7 +115,8 @@ public:
 
     boost::optional<Document> lookupSingleDocument(const NamespaceString& nss,
                                                    UUID collectionUUID,
-                                                   const Document& documentKey) {
+                                                   const Document& documentKey,
+                                                   boost::optional<BSONObj> readConcern) {
         boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest(nss));
         auto swPipeline = makePipeline({BSON("$match" << documentKey)}, expCtx);
         if (swPipeline == ErrorCodes::NamespaceNotFound) {
diff --git a/src/mongo/db/pipeline/expression_context.cpp b/src/mongo/db/pipeline/expression_context.cpp
index e2b2e1e758d..471a66626e1 100644
--- a/src/mongo/db/pipeline/expression_context.cpp
+++ b/src/mongo/db/pipeline/expression_context.cpp
@@ -46,6 +46,7 @@ ExpressionContext::ExpressionContext(OperationContext* opCtx,
                                      StringMap<ResolvedNamespace> resolvedNamespaces)
     : ExpressionContext(opCtx, collator.get()) {
     explain = request.getExplain();
+    comment = request.getComment();
     fromMongos = request.isFromMongos();
     needsMerge = request.needsMerge();
     allowDiskUse = request.shouldAllowDiskUse();
@@ -131,6 +132,7 @@ intrusive_ptr<ExpressionContext> ExpressionContext::copyWith(
 
     expCtx->uuid = std::move(uuid);
     expCtx->explain = explain;
+    expCtx->comment = comment;
     expCtx->needsMerge = needsMerge;
     expCtx->fromMongos = fromMongos;
     expCtx->from34Mongos = from34Mongos;
diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index a2e16163a73..e949e068f2b 100644
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -165,6 +165,9 @@ public:
     // The explain verbosity requested by the user, or boost::none if no explain was requested.
     boost::optional<ExplainOptions::Verbosity> explain;
 
+    // The comment provided by the user, or the empty string if no comment was provided.
+    std::string comment;
+
     bool fromMongos = false;
     bool needsMerge = false;
     bool inMongos = false;
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index b32f18828d0..f19c9aa3abf 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -386,12 +386,15 @@ public:
 
     boost::optional<Document> lookupSingleDocument(const NamespaceString& nss,
                                                    UUID collectionUUID,
-                                                   const Document& documentKey) final {
+                                                   const Document& documentKey,
+                                                   boost::optional<BSONObj> readConcern) final {
+        invariant(!readConcern);  // We don't currently support a read concern on mongod - it's only
+                                  // expected to be necessary on mongos.
+                                  //
         // Be sure to do the lookup using the collection default collation.
         auto foreignExpCtx =
             _ctx->copyWith(nss, collectionUUID, _getCollectionDefaultCollator(nss, collectionUUID));
         auto swPipeline = makePipeline({BSON("$match" << documentKey)}, foreignExpCtx);
-
         if (swPipeline == ErrorCodes::NamespaceNotFound) {
             return boost::none;
         }
diff --git a/src/mongo/db/pipeline/stub_mongo_process_interface.h b/src/mongo/db/pipeline/stub_mongo_process_interface.h
index c0beb0504ba..6b0032b1fd3 100644
--- a/src/mongo/db/pipeline/stub_mongo_process_interface.h
+++ b/src/mongo/db/pipeline/stub_mongo_process_interface.h
@@ -120,7 +120,8 @@ public:
 
     boost::optional<Document> lookupSingleDocument(const NamespaceString& nss,
                                                    UUID collectionUUID,
-                                                   const Document& documentKey) {
+                                                   const Document& documentKey,
+                                                   boost::optional<BSONObj> readConcern) {
         MONGO_UNREACHABLE;
     }
 
diff --git a/src/mongo/s/commands/pipeline_s.cpp b/src/mongo/s/commands/pipeline_s.cpp
index b9fef455894..ef75292be37 100644
--- a/src/mongo/s/commands/pipeline_s.cpp
+++ b/src/mongo/s/commands/pipeline_s.cpp
@@ -32,6 +32,7 @@
 
 #include "mongo/db/pipeline/document_source.h"
 #include "mongo/db/query/collation/collation_spec.h"
+#include "mongo/db/repl/read_concern_args.h"
 #include "mongo/executor/task_executor_pool.h"
 #include "mongo/s/catalog_cache.h"
 #include "mongo/s/commands/cluster_commands_helpers.h"
@@ -175,7 +176,8 @@ public:
 
     boost::optional<Document> lookupSingleDocument(const NamespaceString& nss,
                                                    UUID collectionUUID,
-                                                   const Document& filter) final {
+                                                   const Document& filter,
+                                                   boost::optional<BSONObj> readConcern) final {
         auto foreignExpCtx = _expCtx->copyWith(nss, collectionUUID);
 
         // Create the find command to be dispatched to the shard in order to return the post-change
@@ -189,12 +191,16 @@ public:
             cmdBuilder.append("find", nss.coll());
         }
         cmdBuilder.append("filter", filterObj);
+        cmdBuilder.append("comment", _expCtx->comment);
+        if (readConcern) {
+            cmdBuilder.append(repl::ReadConcernArgs::kReadConcernFieldName, *readConcern);
+        }
 
         auto swShardResult = makeStatusWith<std::vector<ClusterClientCursorParams::RemoteCursor>>();
         auto findCmd = cmdBuilder.obj();
         size_t numAttempts = 0;
         do {
-            // Verify that the collection exists, with the UUID passed in the expCtx.
+            // Verify that the collection exists, with the correct UUID.
             auto catalogCache = Grid::get(_expCtx->opCtx)->catalogCache();
             auto swRoutingInfo = getCollectionRoutingInfo(foreignExpCtx);
             if (swRoutingInfo == ErrorCodes::NamespaceNotFound) {
diff --git a/src/mongo/s/query/cluster_client_cursor.h b/src/mongo/s/query/cluster_client_cursor.h
index a263a77f56e..742d294f107 100644
--- a/src/mongo/s/query/cluster_client_cursor.h
+++ b/src/mongo/s/query/cluster_client_cursor.h
@@ -30,6 +30,7 @@
 
 #include <boost/optional.hpp>
 
+#include "mongo/client/read_preference.h"
 #include "mongo/db/auth/user_name.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/logical_session_id.h"
@@ -137,6 +138,11 @@ public:
      * Returns the logical session id for this cursor.
      */
     virtual boost::optional<LogicalSessionId> getLsid() const = 0;
+
+    /**
+     * Returns the readPreference for this cursor.
+     */
+    virtual boost::optional<ReadPreferenceSetting> getReadPreference() const = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/query/cluster_client_cursor_impl.cpp b/src/mongo/s/query/cluster_client_cursor_impl.cpp
index 1cad6ef0830..9a2d5de4444 100644
--- a/src/mongo/s/query/cluster_client_cursor_impl.cpp
+++ b/src/mongo/s/query/cluster_client_cursor_impl.cpp
@@ -146,6 +146,10 @@ boost::optional<LogicalSessionId> ClusterClientCursorImpl::getLsid() const {
     return _lsid;
 }
 
+boost::optional<ReadPreferenceSetting> ClusterClientCursorImpl::getReadPreference() const {
+    return _params.readPreference;
+}
+
 namespace {
 
 /**
diff --git a/src/mongo/s/query/cluster_client_cursor_impl.h b/src/mongo/s/query/cluster_client_cursor_impl.h
index f325be6ca4d..67e334a7f47 100644
--- a/src/mongo/s/query/cluster_client_cursor_impl.h
+++ b/src/mongo/s/query/cluster_client_cursor_impl.h
@@ -113,6 +113,8 @@ public:
 
     boost::optional<LogicalSessionId> getLsid() const final;
 
+    boost::optional<ReadPreferenceSetting> getReadPreference() const final;
+
 public:
     /** private for tests */
     /**
diff --git a/src/mongo/s/query/cluster_client_cursor_mock.cpp b/src/mongo/s/query/cluster_client_cursor_mock.cpp
index 4899e0f3056..f2327f861fa 100644
--- a/src/mongo/s/query/cluster_client_cursor_mock.cpp
+++ b/src/mongo/s/query/cluster_client_cursor_mock.cpp
@@ -115,4 +115,8 @@ boost::optional<LogicalSessionId> ClusterClientCursorMock::getLsid() const {
     return _lsid;
 }
 
+boost::optional<ReadPreferenceSetting> ClusterClientCursorMock::getReadPreference() const {
+    return boost::none;
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/query/cluster_client_cursor_mock.h b/src/mongo/s/query/cluster_client_cursor_mock.h
index 5cc6b2149a8..c63a2654cda 100644
--- a/src/mongo/s/query/cluster_client_cursor_mock.h
+++ b/src/mongo/s/query/cluster_client_cursor_mock.h
@@ -69,6 +69,8 @@ public:
 
     boost::optional<LogicalSessionId> getLsid() const final;
 
+    boost::optional<ReadPreferenceSetting> getReadPreference() const final;
+
     /**
      * Returns true unless marked as having non-exhausted remote cursors via
      * markRemotesNotExhausted().
diff --git a/src/mongo/s/query/cluster_cursor_manager.cpp b/src/mongo/s/query/cluster_cursor_manager.cpp
index ed34635a8fc..fd7927e90ea 100644
--- a/src/mongo/s/query/cluster_cursor_manager.cpp
+++ b/src/mongo/s/query/cluster_cursor_manager.cpp
@@ -139,6 +139,12 @@ bool ClusterCursorManager::PinnedCursor::isTailableAndAwaitData() const {
     return _cursor->isTailableAndAwaitData();
 }
 
+boost::optional<ReadPreferenceSetting> ClusterCursorManager::PinnedCursor::getReadPreference()
+    const {
+    invariant(_cursor);
+    return _cursor->getReadPreference();
+}
+
 UserNameIterator ClusterCursorManager::PinnedCursor::getAuthenticatedUsers() const {
     invariant(_cursor);
     return _cursor->getAuthenticatedUsers();
diff --git a/src/mongo/s/query/cluster_cursor_manager.h b/src/mongo/s/query/cluster_cursor_manager.h
index 6bd9e20c3a6..d82e1727e15 100644
--- a/src/mongo/s/query/cluster_cursor_manager.h
+++ b/src/mongo/s/query/cluster_cursor_manager.h
@@ -206,6 +206,11 @@ public:
         CursorId getCursorId() const;
 
         /**
+         * Returns the read preference setting for this cursor.
+         */
+        boost::optional<ReadPreferenceSetting> getReadPreference() const;
+
+        /**
          * Returns the number of result documents returned so far by this cursor via the next()
          * method.
          */
diff --git a/src/mongo/s/query/cluster_find.cpp b/src/mongo/s/query/cluster_find.cpp
index 589917cd46b..ac055b83fd4 100644
--- a/src/mongo/s/query/cluster_find.cpp
+++ b/src/mongo/s/query/cluster_find.cpp
@@ -422,6 +422,9 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx,
                               << " was not created by the authenticated user"};
     }
 
+    if (auto readPref = pinnedCursor.getValue().getReadPreference()) {
+        ReadPreferenceSetting::get(opCtx) = *readPref;
+    }
     if (pinnedCursor.getValue().isTailableAndAwaitData()) {
         // Default to 1-second timeout for tailable awaitData cursors. If an explicit maxTimeMS has
         // been specified, do not apply it to the opCtx, since its deadline will already have been