SERVER-14498 Add DeleteStage, rewrite DeleteExecutor to use it

5 files changed, 324 insertions, 1 deletions

diff --git a/src/mongo/db/exec/SConscript b/src/mongo/db/exec/SConscript
index abfd4c0c458..d3bea2e5974 100644
--- a/src/mongo/db/exec/SConscript
+++ b/src/mongo/db/exec/SConscript
@@ -40,6 +40,7 @@ env.Library(
         "cached_plan.cpp",
         "collection_scan.cpp",
         "count.cpp",
+        "delete.cpp",
         "distinct_scan.cpp",
         "eof.cpp",
         "fetch.cpp",
diff --git a/src/mongo/db/exec/delete.cpp b/src/mongo/db/exec/delete.cpp
new file mode 100644
index 00000000000..f06fac15398
--- /dev/null
+++ b/src/mongo/db/exec/delete.cpp
@@ -0,0 +1,182 @@
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/exec/delete.h"
+
+#include "mongo/db/catalog/collection.h"
+#include "mongo/db/exec/working_set_common.h"
+#include "mongo/db/repl/oplog.h"
+#include "mongo/db/repl/repl_coordinator_global.h"
+
+namespace mongo {
+
+    // static
+    const char* DeleteStage::kStageType = "DELETE";
+
+    DeleteStage::DeleteStage(OperationContext* txn,
+                             const DeleteStageParams& params,
+                             WorkingSet* ws,
+                             Collection* collection,
+                             PlanStage* child)
+        : _txn(txn),
+          _params(params),
+          _ws(ws),
+          _collection(collection),
+          _child(child),
+          _commonStats(kStageType) { }
+
+    DeleteStage::~DeleteStage() {}
+
+    bool DeleteStage::isEOF() {
+        if (!_collection) {
+            return true;
+        }
+        if (!_params.isMulti && _specificStats.docsDeleted > 0) {
+            return true;
+        }
+        return _child->isEOF();
+    }
+
+    PlanStage::StageState DeleteStage::work(WorkingSetID* out) {
+        ++_commonStats.works;
+
+        // Adds the amount of time taken by work() to executionTimeMillis.
+        ScopedTimer timer(&_commonStats.executionTimeMillis);
+
+        if (isEOF()) { return PlanStage::IS_EOF; }
+        invariant(_collection); // If isEOF() returns false, we must have a collection.
+
+        WorkingSetID id = WorkingSet::INVALID_ID;
+        StageState status = _child->work(&id);
+
+        if (PlanStage::ADVANCED == status) {
+            WorkingSetMember* member = _ws->get(id);
+            if (!member->hasLoc()) {
+                _ws->free(id);
+                const std::string errmsg = "delete stage failed to read member w/ loc from child";
+                *out = WorkingSetCommon::allocateStatusMember(_ws, Status(ErrorCodes::InternalError,
+                                                                          errmsg));
+                return PlanStage::FAILURE;
+            }
+            DiskLoc rloc = member->loc;
+            _ws->free(id);
+
+            BSONObj deletedDoc;
+
+            WriteUnitOfWork wunit(_txn->recoveryUnit());
+
+            // TODO: Do we want to buffer docs and delete them in a group rather than
+            // saving/restoring state repeatedly?
+            saveState();
+            const bool deleteCappedOK = false;
+            const bool deleteNoWarn = false;
+            _collection->deleteDocument(_txn, rloc, deleteCappedOK, deleteNoWarn,
+                                        _params.shouldCallLogOp ? &deletedDoc : NULL);
+            restoreState(_txn);
+
+            ++_specificStats.docsDeleted;
+
+            if (_params.shouldCallLogOp) {
+                if (deletedDoc.isEmpty()) {
+                    log() << "Deleted object without id in collection " << _collection->ns()
+                          << ", not logging.";
+                }
+                else {
+                    bool replJustOne = true;
+                    repl::logOp(_txn, "d", _collection->ns().ns().c_str(), deletedDoc, 0,
+                                &replJustOne);
+                }
+            }
+
+            wunit.commit();
+
+            _txn->recoveryUnit()->commitIfNeeded();
+
+            ++_commonStats.needTime;
+            return PlanStage::NEED_TIME;
+        }
+        else if (PlanStage::FAILURE == status) {
+            *out = id;
+            // If a stage fails, it may create a status WSM to indicate why it failed, in which case
+            // 'id' is valid.  If ID is invalid, we create our own error message.
+            if (WorkingSet::INVALID_ID == id) {
+                const std::string errmsg = "delete stage failed to read in results from child";
+                *out = WorkingSetCommon::allocateStatusMember(_ws, Status(ErrorCodes::InternalError,
+                                                                          errmsg));
+                return PlanStage::FAILURE;
+            }
+            return status;
+        }
+        else {
+            if (PlanStage::NEED_TIME == status) {
+                ++_commonStats.needTime;
+            }
+            return status;
+        }
+    }
+
+    void DeleteStage::saveState() {
+        ++_commonStats.yields;
+        _child->saveState();
+    }
+
+    void DeleteStage::restoreState(OperationContext* opCtx) {
+        ++_commonStats.unyields;
+        _child->restoreState(opCtx);
+    }
+
+    void DeleteStage::invalidate(const DiskLoc& dl, InvalidationType type) {
+        ++_commonStats.invalidates;
+        _child->invalidate(dl, type);
+    }
+
+    vector<PlanStage*> DeleteStage::getChildren() const {
+        vector<PlanStage*> children;
+        children.push_back(_child.get());
+        return children;
+    }
+
+    PlanStageStats* DeleteStage::getStats() {
+        _commonStats.isEOF = isEOF();
+        auto_ptr<PlanStageStats> ret(new PlanStageStats(_commonStats, STAGE_DELETE));
+        ret->specific.reset(new DeleteStats(_specificStats));
+        ret->children.push_back(_child->getStats());
+        return ret.release();
+    }
+
+    const CommonStats* DeleteStage::getCommonStats() {
+        return &_commonStats;
+    }
+
+    const SpecificStats* DeleteStage::getSpecificStats() {
+        return &_specificStats;
+    }
+
+}  // namespace mongo
diff --git a/src/mongo/db/exec/delete.h b/src/mongo/db/exec/delete.h
new file mode 100644
index 00000000000..3e24375a5dc
--- /dev/null
+++ b/src/mongo/db/exec/delete.h
@@ -0,0 +1,106 @@
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/plan_stage.h"
+#include "mongo/db/jsobj.h"
+
+namespace mongo {
+
+    class OperationContext;
+
+    struct DeleteStageParams {
+        DeleteStageParams() : isMulti(false), shouldCallLogOp(false) { }
+
+        // Should we delete all documents returned from the child (a "multi delete"), or at most one
+        // (a "single delete")?
+        bool isMulti;
+
+        // Should we write each delete to the oplog?
+        bool shouldCallLogOp;
+    };
+
+    /**
+     * This stage delete documents by DiskLoc that are returned from its child.  NEED_TIME
+     * is returned after deleting a document.
+     *
+     * Callers of work() must be holding a write lock (and, for shouldCallLogOp=true deletes,
+     * callers must have had the replication coordinator approve the write).
+     */
+    class DeleteStage : public PlanStage {
+        MONGO_DISALLOW_COPYING(DeleteStage);
+    public:
+        DeleteStage(OperationContext* txn,
+                    const DeleteStageParams& params,
+                    WorkingSet* ws,
+                    Collection* collection,
+                    PlanStage* child);
+        virtual ~DeleteStage();
+
+        virtual bool isEOF();
+        virtual StageState work(WorkingSetID* out);
+
+        virtual void saveState();
+        virtual void restoreState(OperationContext* opCtx);
+        virtual void invalidate(const DiskLoc& dl, InvalidationType type);
+
+        virtual std::vector<PlanStage*> getChildren() const;
+
+        virtual StageType stageType() const { return STAGE_DELETE; }
+
+        virtual PlanStageStats* getStats();
+
+        virtual const CommonStats* getCommonStats();
+
+        virtual const SpecificStats* getSpecificStats();
+
+        static const char* kStageType;
+
+    private:
+        // Transactional context.  Not owned by us.
+        OperationContext* _txn;
+
+        DeleteStageParams _params;
+
+        // Not owned by us.
+        WorkingSet* _ws;
+
+        // Collection to operate on.  Not owned by us.  Can be NULL (if NULL, isEOF() will always
+        // return true).  If non-NULL, the lifetime of the collection must supersede that of the
+        // stage.
+        Collection* _collection;
+
+        scoped_ptr<PlanStage> _child;
+
+        // Stats
+        CommonStats _commonStats;
+        DeleteStats _specificStats;
+    };
+
+}  // namespace mongo
diff --git a/src/mongo/db/exec/plan_stats.h b/src/mongo/db/exec/plan_stats.h
index f22b9bef7c2..adbc0b6d986 100644
--- a/src/mongo/db/exec/plan_stats.h
+++ b/src/mongo/db/exec/plan_stats.h
@@ -264,6 +264,16 @@ namespace mongo {
 
     };
 
+    struct DeleteStats : public SpecificStats {
+        DeleteStats() : docsDeleted(0) { }
+
+        virtual SpecificStats* clone() const {
+            return new DeleteStats(*this);
+        }
+
+        size_t docsDeleted;
+    };
+
     struct DistinctScanStats : public SpecificStats {
         DistinctScanStats() : keysExamined(0) { }
 
diff --git a/src/mongo/db/exec/stagedebug_cmd.cpp b/src/mongo/db/exec/stagedebug_cmd.cpp
index ea8fff78edd..6251076f8a8 100644
--- a/src/mongo/db/exec/stagedebug_cmd.cpp
+++ b/src/mongo/db/exec/stagedebug_cmd.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/exec/and_hash.h"
 #include "mongo/db/exec/and_sorted.h"
 #include "mongo/db/exec/collection_scan.h"
+#include "mongo/db/exec/delete.h"
 #include "mongo/db/exec/fetch.h"
 #include "mongo/db/exec/index_scan.h"
 #include "mongo/db/exec/limit.h"
@@ -78,6 +79,7 @@ namespace mongo {
      * node -> {skip: {args: {node: node, num: posint}}}
      * node -> {sort: {args: {node: node, pattern: objWithSortCriterion }}}
      * node -> {mergeSort: {args: {nodes: [node, node], pattern: objWithSortCriterion}}}
+     * node -> {delete: {args: {node: node, isMulti: bool, shouldCallLogOp: bool}}}
      *
      * Forthcoming Nodes:
      *
@@ -116,7 +118,10 @@ namespace mongo {
             string collName = collElt.String();
 
             // Need a context to get the actual Collection*
-            Client::ReadContext ctx(txn, dbname);
+            // TODO A write lock is currently taken here to accommodate stages that perform writes
+            //      (e.g. DeleteStage).  This should be changed to use a read lock for read-only
+            //      execution trees.
+            Client::WriteContext ctx(txn, dbname);
 
             // Make sure the collection is valid.
             Database* db = ctx.ctx().db();
@@ -409,6 +414,25 @@ namespace mongo {
 
                 return new TextStage(txn, params, workingSet, matcher);
             }
+            else if ("delete" == nodeName) {
+                uassert(18636, "Delete stage doesn't have a filter (put it on the child)",
+                        NULL == matcher);
+                uassert(18637, "node argument must be provided to delete",
+                        nodeArgs["node"].isABSONObj());
+                uassert(18638, "isMulti argument must be provided to delete",
+                        nodeArgs["isMulti"].type() == Bool);
+                uassert(18639, "shouldCallLogOp argument must be provided to delete",
+                        nodeArgs["shouldCallLogOp"].type() == Bool);
+                PlanStage* subNode = parseQuery(txn,
+                                                collection,
+                                                nodeArgs["node"].Obj(),
+                                                workingSet,
+                                                exprs);
+                DeleteStageParams params;
+                params.isMulti = nodeArgs["isMulti"].Bool();
+                params.shouldCallLogOp = nodeArgs["shouldCallLogOp"].Bool();
+                return new DeleteStage(txn, params, workingSet, collection, subNode);
+            }
             else {
                 return NULL;
             }