SERVER-14498 Add DeleteStage, rewrite DeleteExecutor to use it

1 files changed, 21 insertions, 76 deletions

diff --git a/src/mongo/db/ops/delete_executor.cpp b/src/mongo/db/ops/delete_executor.cpp
index 42459e77f05..ba36650e634 100644
--- a/src/mongo/db/ops/delete_executor.cpp
+++ b/src/mongo/db/ops/delete_executor.cpp
@@ -32,15 +32,11 @@
 
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/database.h"
-#include "mongo/db/client.h"
-#include "mongo/db/curop.h"
+#include "mongo/db/exec/delete.h"
 #include "mongo/db/ops/delete_request.h"
 #include "mongo/db/query/canonical_query.h"
 #include "mongo/db/query/get_executor.h"
-#include "mongo/db/query/lite_parsed_query.h"
-#include "mongo/db/query/query_planner_common.h"
 #include "mongo/db/repl/repl_coordinator_global.h"
-#include "mongo/db/repl/oplog.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/mongoutils/str.h"
 
@@ -87,7 +83,7 @@ namespace mongo {
                 mongoutils::str::stream() <<
                 "DeleteExecutor::prepare() failed to parse query " << _request->getQuery(),
                 _isQueryParsed);
-        const bool logop = _request->shouldCallLogOp();
+
         const NamespaceString& ns(_request->getNamespaceString());
         if (!_request->isGod()) {
             if (ns.isSystem()) {
@@ -97,7 +93,7 @@ namespace mongo {
             }
             if (ns.ns().find('$') != string::npos) {
                 log() << "cannot delete from collection with reserved $ in name: " << ns << endl;
-                uasserted( 10100, "cannot delete from collection with reserved $ in name" );
+                uasserted(10100, "cannot delete from collection with reserved $ in name");
             }
         }
 
@@ -112,86 +108,35 @@ namespace mongo {
 
         uassert(ErrorCodes::NotMaster,
                 str::stream() << "Not primary while removing from " << ns.ns(),
-                !logop ||
+                !_request->shouldCallLogOp() ||
                 repl::getGlobalReplicationCoordinator()->canAcceptWritesForDatabase(ns.db()));
 
-        long long nDeleted = 0;
-
         PlanExecutor* rawExec;
         if (_canonicalQuery.get()) {
-            uassertStatusOK(getExecutor(_request->getOpCtx(),
-                                        collection,
-                                        _canonicalQuery.release(),
-                                        &rawExec));
+            // This is the non-idhack branch.
+            uassertStatusOK(getExecutorDelete(_request->getOpCtx(), collection,
+                                              _canonicalQuery.release(), _request->isMulti(),
+                                              _request->shouldCallLogOp(), &rawExec));
         }
         else {
-            uassertStatusOK(getExecutor(_request->getOpCtx(),
-                                        collection,
-                                        ns.ns(),
-                                        _request->getQuery(),
-                                        &rawExec));
+            // This is the idhack branch.
+            uassertStatusOK(getExecutorDelete(_request->getOpCtx(), collection, ns.ns(),
+                                              _request->getQuery(), _request->isMulti(),
+                                              _request->shouldCallLogOp(), &rawExec));
         }
-
-        auto_ptr<PlanExecutor> exec(rawExec);
+        scoped_ptr<PlanExecutor> exec(rawExec);
 
         // Concurrently mutating state (by us) so we need to register 'exec'.
-        ScopedExecutorRegistration safety(exec.get());
-
-        DiskLoc rloc;
-        PlanExecutor::ExecState state;
-        CurOp* curOp = _request->getOpCtx()->getCurOp();
-        int oldYieldCount = curOp->numYields();
-        while (PlanExecutor::ADVANCED == (state = exec->getNext(NULL, &rloc))) {
-            if (oldYieldCount != curOp->numYields()) {
-                uassert(ErrorCodes::NotMaster,
-                        str::stream() << "No longer primary while removing from " << ns.ns(),
-                        !logop ||
-                        repl::getGlobalReplicationCoordinator()->canAcceptWritesForDatabase(
-                                ns.db()));
-                oldYieldCount = curOp->numYields();
-            }
-            BSONObj toDelete;
-
-            WriteUnitOfWork wunit(_request->getOpCtx()->recoveryUnit());
-
-            // TODO: do we want to buffer docs and delete them in a group rather than
-            // saving/restoring state repeatedly?
-            exec->saveState();
-            collection->deleteDocument(
-                            _request->getOpCtx(), rloc, false, false, logop ? &toDelete : NULL);
-            exec->restoreState(_request->getOpCtx());
-
-            nDeleted++;
-
-            if (logop) {
-                if ( toDelete.isEmpty() ) {
-                    log() << "Deleted object without id in collection " << collection->ns()
-                          << ", not logging.";
-                }
-                else {
-                    bool replJustOne = true;
-                    repl::logOp(
-                            _request->getOpCtx(), "d", ns.ns().c_str(), toDelete, 0, &replJustOne);
-                }
-            }
-
-            wunit.commit();
+        const ScopedExecutorRegistration safety(exec.get());
 
-            if (!_request->isMulti()) {
-                break;
-            }
-
-            if (!_request->isGod()) {
-                _request->getOpCtx()->recoveryUnit()->commitIfNeeded();
-            }
-
-            if (debug && _request->isGod() && nDeleted == 100) {
-                log() << "warning high number of deletes with god=true "
-                      << " which could use significant memory b/c we don't commit journal";
-            }
-        }
+        uassertStatusOK(exec->executePlan());
 
-        return nDeleted;
+        // Extract the number of documents deleted from the DeleteStage stats.
+        invariant(exec->getRootStage()->stageType() == STAGE_DELETE);
+        DeleteStage* deleteStage = static_cast<DeleteStage*>(exec->getRootStage());
+        const DeleteStats* deleteStats =
+            static_cast<const DeleteStats*>(deleteStage->getSpecificStats());
+        return deleteStats->docsDeleted;
     }
 
 }  // namespace mongo