SERVER-42181 Make WorkingSetMember hold Document instead of BSONObj

SERVER-42157 Unowned mode for Document/Value

Author: Martin Neupauer

src/mongo/db/commands/list_collections.cpp

@@ -131,7 +131,7 @@ void _addWorkingSetMember(OperationContext* opCtx,
     WorkingSetMember* member = ws->get(id);
     member->keyData.clear();
     member->recordId = RecordId();
-    member->obj = Snapshotted<BSONObj>(SnapshotId(), maybe);
+    member->resetDocument(SnapshotId(), maybe);
     member->transitionToOwnedObj();
     root->pushBack(id);
 }

src/mongo/db/commands/list_indexes.cpp

@@ -190,7 +190,7 @@ class CmdListIndexes : public BasicCommand {
                 WorkingSetMember* member = ws->get(id);
                 member->keyData.clear();
                 member->recordId = RecordId();
-                member->obj = Snapshotted<BSONObj>(SnapshotId(), indexSpec.getOwned());
+                member->resetDocument(SnapshotId(), indexSpec.getOwned());
                 member->transitionToOwnedObj();
                 root->pushBack(id);
             }

src/mongo/db/exec/and_common.h

@@ -64,7 +64,7 @@ class AndCommon {
             invariant(src.getState() == WorkingSetMember::RID_AND_OBJ);
 
             // 'src' has the full document but 'dest' doesn't so we need to copy it over.
-            dest->obj = src.obj;
+            dest->doc = src.doc;
             dest->makeObjOwnedIfNeeded();
 
             // We have an object so we don't need key data.

src/mongo/db/exec/cached_plan.cpp

@@ -139,8 +139,7 @@ Status CachedPlanStage::pickBestPlan(PlanYieldPolicy* yieldPolicy) {
         } else if (PlanStage::FAILURE == state) {
             // On failure, fall back to replanning the whole query. We neither evict the
             // existing cache entry nor cache the result of replanning.
-            BSONObj statusObj;
-            WorkingSetCommon::getStatusMemberObject(*_ws, id, &statusObj);
+            BSONObj statusObj = WorkingSetCommon::getStatusMemberDocument(*_ws, id)->toBson();
 
             LOG(1) << "Execution of cached plan failed, falling back to replan."
                    << " query: " << redact(_canonicalQuery->toStringShort())

src/mongo/db/exec/change_stream_proxy.cpp

@@ -50,15 +50,15 @@ ChangeStreamProxyStage::ChangeStreamProxyStage(OperationContext* opCtx,
     _latestOplogTimestamp = ResumeToken::parse(_postBatchResumeToken).getData().clusterTime;
 }
 
-boost::optional<BSONObj> ChangeStreamProxyStage::getNextBson() {
+boost::optional<Document> ChangeStreamProxyStage::getNext() {
     if (auto next = _pipeline->getNext()) {
         // While we have more results to return, we track both the timestamp and the resume token of
         // the latest event observed in the oplog, the latter via its sort key metadata field.
-        auto nextBSON = _validateAndConvertToBSON(*next);
+        _validateResumeToken(*next);
         _latestOplogTimestamp = PipelineD::getLatestOplogTimestamp(_pipeline.get());
         _postBatchResumeToken = next->metadata().getSortKey();
         _setSpeculativeReadTimestamp();
-        return nextBSON;
+        return next;
     }
 
     // We ran out of results to return. Check whether the oplog cursor has moved forward since the
@@ -76,10 +76,10 @@ boost::optional<BSONObj> ChangeStreamProxyStage::getNextBson() {
     return boost::none;
 }
 
-BSONObj ChangeStreamProxyStage::_validateAndConvertToBSON(const Document& event) const {
+void ChangeStreamProxyStage::_validateResumeToken(const Document& event) const {
     // If we are producing output to be merged on mongoS, then no stages can have modified the _id.
     if (_includeMetaData) {
-        return event.toBsonWithMetaData();
+        return;
     }
     // Confirm that the document _id field matches the original resume token in the sort key field.
     auto eventBSON = event.toBson();
@@ -95,7 +95,6 @@ BSONObj ChangeStreamProxyStage::_validateAndConvertToBSON(const Document& event)
                           << BSON("_id" << resumeToken) << " but found: "
                           << (eventBSON["_id"] ? BSON("_id" << eventBSON["_id"]) : BSONObj()),
             idField.binaryEqual(resumeToken));
-    return eventBSON;
 }
 
 void ChangeStreamProxyStage::_setSpeculativeReadTimestamp() {

src/mongo/db/exec/change_stream_proxy.h

@@ -77,13 +77,13 @@ class ChangeStreamProxyStage final : public PipelineProxyStage {
     }
 
 protected:
-    boost::optional<BSONObj> getNextBson() final;
+    boost::optional<Document> getNext() final;
 
 private:
     /**
-     * Verifies that the docs's resume token has not been modified, then converts the doc to BSON.
+     * Verifies that the docs's resume token has not been modified.
      */
-    BSONObj _validateAndConvertToBSON(const Document& event) const;
+    void _validateResumeToken(const Document& event) const;
 
     /**
      * Set the speculative majority read timestamp if we have scanned up to a certain oplog

src/mongo/db/exec/collection_scan.cpp

@@ -186,7 +186,8 @@ PlanStage::StageState CollectionScan::doWork(WorkingSetID* out) {
     WorkingSetID id = _workingSet->allocate();
     WorkingSetMember* member = _workingSet->get(id);
     member->recordId = record->id;
-    member->obj = {getOpCtx()->recoveryUnit()->getSnapshotId(), record->data.releaseToBson()};
+    member->resetDocument(getOpCtx()->recoveryUnit()->getSnapshotId(),
+                          record->data.releaseToBson());
     _workingSet->transitionToRecordIdAndObj(id);
 
     return returnIfMatches(member, id, out);

src/mongo/db/exec/delete.cpp

@@ -183,12 +183,11 @@ PlanStage::StageState DeleteStage::doWork(WorkingSetID* out) {
     if (_params->returnDeleted) {
         // Save a copy of the document that is about to get deleted, but keep it in the RID_AND_OBJ
         // state in case we need to retry deleting it.
-        BSONObj deletedDoc = member->obj.value();
-        member->obj.setValue(deletedDoc.getOwned());
+        member->makeObjOwnedIfNeeded();
     }
 
     if (_params->removeSaver) {
-        uassertStatusOK(_params->removeSaver->goingToDelete(member->obj.value()));
+        uassertStatusOK(_params->removeSaver->goingToDelete(member->doc.value().toBson()));
     }
 
     // TODO: Do we want to buffer docs and delete them in a group rather than saving/restoring state

src/mongo/db/exec/filter.h

@@ -42,20 +42,20 @@ namespace mongo {
  */
 class WorkingSetMatchableDocument : public MatchableDocument {
 public:
-    WorkingSetMatchableDocument(WorkingSetMember* wsm) : _wsm(wsm) {}
+    WorkingSetMatchableDocument(WorkingSetMember* wsm)
+        : _wsm(wsm), _obj(_wsm->doc.value().toBson()) {}
 
     // This is only called by a $where query.  The query system must be smart enough to realize
     // that it should do a fetch beforehand.
     BSONObj toBSON() const {
-        invariant(_wsm->hasObj());
-        return _wsm->obj.value();
+        return _obj;
     }
 
     ElementIterator* allocateIterator(const ElementPath* path) const final {
         // BSONElementIterator does some interesting things with arrays that I don't think
         // SimpleArrayElementIterator does.
         if (_wsm->hasObj()) {
-            return new BSONElementIterator(path, _wsm->obj.value());
+            return new BSONElementIterator(path, _obj);
         }
 
         // NOTE: This (kind of) duplicates code in WorkingSetMember::getFieldDotted.
@@ -95,6 +95,7 @@ class WorkingSetMatchableDocument : public MatchableDocument {
 
 private:
     WorkingSetMember* _wsm;
+    BSONObj _obj;
 };
 
 class IndexKeyMatchableDocument : public MatchableDocument {

src/mongo/db/exec/geo_near.cpp

@@ -151,7 +151,7 @@ static StatusWith<double> computeGeoNearDistance(const GeoNearParams& nearParams
 
     // Extract all the geometries out of this document for the near query
     std::vector<std::unique_ptr<StoredGeometry>> geometries;
-    extractGeometries(member->obj.value(), nearParams.nearQuery->field, &geometries);
+    extractGeometries(member->doc.value().toBson(), nearParams.nearQuery->field, &geometries);
 
     // Compute the minimum distance of all the geometries in the document
     double minDistance = -1;