SERVER-17303 findAndModify upsert calls Collection::insertDocument() directly

(cherry picked from commit 30d9e17410a3dec85ca2a148c745a6b8f9a8ecd0)

3 files changed, 169 insertions, 61 deletions

diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 911a9c757ed..fea571abd48 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -37,12 +37,14 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/dbhelpers.h"
+#include "mongo/db/exec/update.h"
 #include "mongo/db/exec/working_set_common.h"
 #include "mongo/db/projection.h"
 #include "mongo/db/ops/delete.h"
 #include "mongo/db/ops/update.h"
 #include "mongo/db/ops/update_lifecycle_impl.h"
 #include "mongo/db/query/get_executor.h"
+#include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/util/log.h"
 
@@ -377,17 +379,79 @@ namespace mongo {
             }
             else {
                 // update
-                if ( ! found && ! upsert ) {
-                    // didn't have it, and am not upserting
-                    _appendHelper(result, doc, found, fields, whereCallback);
+                if (!found) {
+                    if (!upsert) {
+                        // Didn't have it, and not upserting.
+                        _appendHelper(result, doc, found, fields, whereCallback);
+                    }
+                    else {
+                        // Do an insert.
+                        BSONObj newDoc;
+                        {
+                            CanonicalQuery* rawCq;
+                            uassertStatusOK(CanonicalQuery::canonicalize(ns, queryModified, &rawCq,
+                                                                         WhereCallbackNoop()));
+                            boost::scoped_ptr<CanonicalQuery> cq(rawCq);
+
+                            UpdateDriver::Options opts;
+                            UpdateDriver driver(opts);
+                            uassertStatusOK(driver.parse(update));
+
+                            mutablebson::Document doc(newDoc,
+                                                      mutablebson::Document::kInPlaceDisabled);
+
+                            const bool ignoreVersion = false;
+                            UpdateLifecycleImpl updateLifecycle(ignoreVersion, collection->ns());
+
+                            UpdateStats stats;
+                            const bool isInternalRequest = false;
+
+                            uassertStatusOK(UpdateStage::applyUpdateOpsForInsert(cq.get(),
+                                                                                 queryModified,
+                                                                                 &driver,
+                                                                                 &updateLifecycle,
+                                                                                 &doc,
+                                                                                 isInternalRequest,
+                                                                                 &stats,
+                                                                                 &newDoc));
+                        }
+
+                        const bool enforceQuota = true;
+                        uassertStatusOK(collection->insertDocument(txn, newDoc, enforceQuota)
+                                        .getStatus());
+
+                        // This is the last thing we do before the WriteUnitOfWork commits (except
+                        // for some BSON manipulation).
+                        repl::logOp(txn, "i", collection->ns().ns().c_str(), newDoc);
+
+                        if (returnNew) {
+                            // The third argument, set to true here, indicates whether or not we
+                            // have something for the 'value' field returned by a findAndModify
+                            // command.
+                            //
+                            // If we're returning the old version of the document, then the this
+                            // boolean is set based on whether or not we found something.
+                            //
+                            // Here we didn't find a document, but we inserted a document and the
+                            // user is asking for the new doc back. Therefore, we always have a
+                            // value to return, so we just pass true.
+                            _appendHelper(result, newDoc, true, fields, whereCallback);
+                        }
+
+                        BSONObjBuilder le(result.subobjStart("lastErrorObject"));
+                        le.appendBool("updatedExisting", false);
+                        le.appendNumber("n", 1);
+                        le.appendAs(newDoc["_id"], kUpsertedFieldName);
+                        le.done();
+                    }
                 }
                 else {
                     // we found it or we're updating
-                    
+
                     if ( ! returnNew ) {
                         _appendHelper(result, doc, found, fields, whereCallback);
                     }
-                    
+
                     const NamespaceString requestNs(ns);
                     UpdateRequest request(requestNs);
 
@@ -408,21 +472,10 @@ namespace mongo {
                                                      request,
                                                      &txn->getCurOp()->debug());
 
-                    if (!found && res.existing) {
-                        // No match was found during the read part of this find and modify, which
-                        // means that we're here doing an upsert. But the update also told us that
-                        // we modified an *already existing* document. This probably means that
-                        // the query reported EOF based on an out-of-date snapshot. This should be
-                        // a rare event, so we handle it by throwing a write conflict.
-                        throw WriteConflictException();
-                    }
-
-                    if ( !collection ) {
-                        // collection created by an upsert
-                        collection = ctx.db()->getCollection(ns);
-                    }
+                    invariant(collection);
+                    invariant(res.existing);
+                    LOG(3) << "update result: "  << res;
 
-                    LOG(3) << "update result: "  << res ;
                     if (returnNew) {
                         dassert(!res.newObj.isEmpty());
                         _appendHelper(result, res.newObj, true, fields, whereCallback);
diff --git a/src/mongo/db/exec/update.cpp b/src/mongo/db/exec/update.cpp
index 329901bfe77..99867cffcfd 100644
--- a/src/mongo/db/exec/update.cpp
+++ b/src/mongo/db/exec/update.cpp
@@ -423,8 +423,8 @@ namespace mongo {
             }
 
             return Status::OK();
-
         }
+
     } // namespace
 
     // static
@@ -621,14 +621,15 @@ namespace mongo {
         }
     }
 
-    void UpdateStage::doInsert() {
-        _specificStats.inserted = true;
-
-        const UpdateRequest* request = _params.request;
-        UpdateDriver* driver = _params.driver;
-        CanonicalQuery* cq = _params.canonicalQuery;
-        UpdateLifecycle* lifecycle = request->getLifecycle();
-
+    // static
+    Status UpdateStage::applyUpdateOpsForInsert(const CanonicalQuery* cq,
+                                                const BSONObj& query,
+                                                UpdateDriver* driver,
+                                                UpdateLifecycle* lifecycle,
+                                                mutablebson::Document* doc,
+                                                bool isInternalRequest,
+                                                UpdateStats* stats,
+                                                BSONObj* out) {
         // Since this is an insert (no docs found and upsert:true), we will be logging it
         // as an insert in the oplog. We don't need the driver's help to build the
         // oplog record, then. We also set the context of the update driver to the INSERT_CONTEXT.
@@ -636,61 +637,88 @@ namespace mongo {
         driver->setLogOp(false);
         driver->setContext(ModifierInterface::ExecInfo::INSERT_CONTEXT);
 
-        // Reset the document we will be writing to
-        _doc.reset();
-
-        // The original document we compare changes to - immutable paths must not change
-        BSONObj original;
-
-        bool isInternalRequest = request->isFromReplication() || request->isFromMigration();
-
         const vector<FieldRef*>* immutablePaths = NULL;
         if (!isInternalRequest && lifecycle)
             immutablePaths = lifecycle->getImmutableFields();
 
-        // Calling populateDocumentWithQueryFields will populate the '_doc' with fields from the
-        // query which creates the base of the update for the inserted doc (because upsert
-        // was true).
+        // The original document we compare changes to - immutable paths must not change
+        BSONObj original;
+
         if (cq) {
-            uassertStatusOK(driver->populateDocumentWithQueryFields(cq, immutablePaths, _doc));
+            Status status = driver->populateDocumentWithQueryFields(cq, immutablePaths, *doc);
+            if (!status.isOK()) {
+                return status;
+            }
+
             if (driver->isDocReplacement())
-                _specificStats.fastmodinsert = true;
-            original = _doc.getObject();
+                stats->fastmodinsert = true;
+            original = doc->getObject();
         }
         else {
-            fassert(17354, CanonicalQuery::isSimpleIdQuery(request->getQuery()));
-            BSONElement idElt = request->getQuery()[idFieldName];
+            fassert(17354, CanonicalQuery::isSimpleIdQuery(query));
+            BSONElement idElt = query[idFieldName];
             original = idElt.wrap();
-            fassert(17352, _doc.root().appendElement(idElt));
+            fassert(17352, doc->root().appendElement(idElt));
         }
 
-        // Apply the update modifications and then log the update as an insert manually.
-        Status status = driver->update(StringData(), &_doc);
-        if (!status.isOK()) {
-            uasserted(16836, status.reason());
+        // Apply the update modifications here.
+        Status updateStatus = driver->update(StringData(), doc);
+        if (!updateStatus.isOK()) {
+            return Status(updateStatus.code(), updateStatus.reason(), 16836);
         }
 
         // Ensure _id exists and is first
-        uassertStatusOK(ensureIdAndFirst(_doc));
+        Status idAndFirstStatus = ensureIdAndFirst(*doc);
+        if (!idAndFirstStatus.isOK()) {
+            return idAndFirstStatus;
+        }
 
         // Validate that the object replacement or modifiers resulted in a document
         // that contains all the immutable keys and can be stored if it isn't coming
         // from a migration or via replication.
-        if (!isInternalRequest){
+        if (!isInternalRequest) {
             FieldRefSet noFields;
             // This will only validate the modified fields if not a replacement.
-            uassertStatusOK(validate(original,
-                                     noFields,
-                                     _doc,
-                                     immutablePaths,
-                                     driver->modOptions()) );
+            Status validateStatus = validate(original,
+                                             noFields,
+                                             *doc,
+                                             immutablePaths,
+                                             driver->modOptions());
+            if (!validateStatus.isOK()) {
+                return validateStatus;
+            }
         }
 
-        // Insert the doc
-        BSONObj newObj = _doc.getObject();
-        uassert(17420,
-                str::stream() << "Document to upsert is larger than " << BSONObjMaxUserSize,
-                newObj.objsize() <= BSONObjMaxUserSize);
+        BSONObj newObj = doc->getObject();
+        if (newObj.objsize() > BSONObjMaxUserSize) {
+            return Status(ErrorCodes::InvalidBSON,
+                          str::stream() << "Document to upsert is larger than "
+                                        << BSONObjMaxUserSize,
+                          17420);
+        }
+
+        *out = newObj;
+        return Status::OK();
+    }
+
+    void UpdateStage::doInsert() {
+        _specificStats.inserted = true;
+
+        const UpdateRequest* request = _params.request;
+        bool isInternalRequest = request->isFromReplication() || request->isFromMigration();
+
+        // Reset the document we will be writing to.
+        _doc.reset();
+
+        BSONObj newObj;
+        uassertStatusOK(applyUpdateOpsForInsert(_params.canonicalQuery,
+                                                request->getQuery(),
+                                                _params.driver,
+                                                request->getLifecycle(),
+                                                &_doc,
+                                                isInternalRequest,
+                                                &_specificStats,
+                                                &newObj));
 
         _specificStats.objInserted = newObj;
         if (request->shouldStoreResultDoc()) {
diff --git a/src/mongo/db/exec/update.h b/src/mongo/db/exec/update.h
index 23505d06bed..0412355ae69 100644
--- a/src/mongo/db/exec/update.h
+++ b/src/mongo/db/exec/update.h
@@ -115,6 +115,33 @@ namespace mongo {
          */
         static UpdateResult makeUpdateResult(PlanExecutor* exec, OpDebug* opDebug);
 
+        /**
+         * Computes the document to insert if the upsert flag is set to true and no matching
+         * documents are found in the database. The document to upsert is computing using the
+         * query 'cq' and the update mods contained in 'driver'.
+         *
+         * If 'cq' is NULL, which can happen for the idhack update fast path, then 'query' is
+         * used to compute the doc to insert instead of 'cq'.
+         *
+         * 'doc' is the mutable BSON document which you would like the update driver to use
+         * when computing the document to insert.
+         *
+         * Set 'isInternalRequest' to true if the upsert was issued by the replication or
+         * sharding systems.
+         *
+         * Fills out whether or not this is a fastmodinsert in 'stats'.
+         *
+         * Returns the document to insert in *out.
+         */
+        static Status applyUpdateOpsForInsert(const CanonicalQuery* cq,
+                                              const BSONObj& query,
+                                              UpdateDriver* driver,
+                                              UpdateLifecycle* lifecycle,
+                                              mutablebson::Document* doc,
+                                              bool isInternalRequest,
+                                              UpdateStats* stats,
+                                              BSONObj* out);
+
     private:
         /**
          * Computes the result of applying mods to the document 'oldObj' at RecordId 'loc' in