Rewrite Document and Value classes

35 files changed, 2110 insertions, 1821 deletions

diff --git a/src/mongo/db/pipeline/accumulator.cpp b/src/mongo/db/pipeline/accumulator.cpp
index 993ec126504..4b610bec63b 100755
--- a/src/mongo/db/pipeline/accumulator.cpp
+++ b/src/mongo/db/pipeline/accumulator.cpp
@@ -36,10 +36,8 @@ namespace mongo {
         ExpressionNary() {
     }
 
-    void Accumulator::opToBson(BSONObjBuilder *pBuilder,
-                               const std::string& opName,
-                               const std::string& fieldName,
-                               bool requireExpression) const {
+    void Accumulator::opToBson(BSONObjBuilder *pBuilder, StringData opName,
+                               StringData fieldName, bool requireExpression) const {
         verify(vpOperand.size() == 1);
         BSONObjBuilder builder;
         vpOperand[0]->addToBsonObj(&builder, opName, requireExpression);
@@ -47,7 +45,7 @@ namespace mongo {
     }
 
     void Accumulator::addToBsonObj(BSONObjBuilder *pBuilder,
-                                   const std::string& fieldName,
+                                   StringData fieldName,
                                    bool requireExpression) const {
         opToBson(pBuilder, getOpName(), fieldName, requireExpression);
     }
diff --git a/src/mongo/db/pipeline/accumulator.h b/src/mongo/db/pipeline/accumulator.h
index 1c947c9ca84..16606a0dbbd 100755
--- a/src/mongo/db/pipeline/accumulator.h
+++ b/src/mongo/db/pipeline/accumulator.h
@@ -31,9 +31,9 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const;
         virtual void addToBsonArray(BSONArrayBuilder *pBuilder) const;
 
         /*
@@ -41,7 +41,7 @@ namespace mongo {
 
           @returns the accumulated value
          */
-        virtual intrusive_ptr<const Value> getValue() const = 0;
+        virtual Value getValue() const = 0;
 
     protected:
         Accumulator();
@@ -55,9 +55,8 @@ namespace mongo {
           @param fieldName the projected name
           @param opName the operator name
          */
-        void opToBson(
-            BSONObjBuilder *pBuilder, const std::string& fieldName, const std::string& opName,
-            bool requireExpression) const;
+        void opToBson(BSONObjBuilder *pBuilder, StringData opName,
+                      StringData fieldName, bool requireExpression) const;
     };
 
 
@@ -65,9 +64,8 @@ namespace mongo {
         public Accumulator {
     public:
         // virtuals from Expression
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual intrusive_ptr<const Value> getValue() const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual Value getValue() const;
         virtual const char *getOpName() const;
 
         /*
@@ -81,7 +79,7 @@ namespace mongo {
 
     private:
         AccumulatorAddToSet(const intrusive_ptr<ExpressionContext> &pTheCtx);
-        typedef boost::unordered_set<intrusive_ptr<const Value>, Value::Hash > SetType;
+        typedef boost::unordered_set<Value, Value::Hash > SetType;
         mutable SetType set;
         mutable SetType::iterator itr; 
         intrusive_ptr<ExpressionContext> pCtx;
@@ -98,12 +96,12 @@ namespace mongo {
         public Accumulator {
     public:
         // virtuals from Expression
-        virtual intrusive_ptr<const Value> getValue() const;
+        virtual Value getValue() const;
 
     protected:
         AccumulatorSingleValue();
 
-        mutable intrusive_ptr<const Value> pValue; /* current min/max */
+        mutable Value pValue; /* current min/max */
     };
 
 
@@ -111,8 +109,7 @@ namespace mongo {
         public AccumulatorSingleValue {
     public:
         // virtuals from Expression
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
 
         /*
@@ -132,8 +129,7 @@ namespace mongo {
         public AccumulatorSingleValue {
     public:
         // virtuals from Expression
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
 
         /*
@@ -153,9 +149,8 @@ namespace mongo {
         public Accumulator {
     public:
         // virtuals from Accumulator
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual intrusive_ptr<const Value> getValue() const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual Value getValue() const;
         virtual const char *getOpName() const;
 
         /*
@@ -182,8 +177,7 @@ namespace mongo {
         public AccumulatorSingleValue {
     public:
         // virtuals from Expression
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
 
         /*
@@ -207,9 +201,8 @@ namespace mongo {
         public Accumulator {
     public:
         // virtuals from Expression
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual intrusive_ptr<const Value> getValue() const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual Value getValue() const;
         virtual const char *getOpName() const;
 
         /*
@@ -224,7 +217,7 @@ namespace mongo {
     private:
         AccumulatorPush(const intrusive_ptr<ExpressionContext> &pTheCtx);
 
-        mutable vector<intrusive_ptr<const Value> > vpValue;
+        mutable vector<Value> vpValue;
         intrusive_ptr<ExpressionContext> pCtx;
     };
 
@@ -234,9 +227,8 @@ namespace mongo {
         typedef AccumulatorSum Super;
     public:
         // virtuals from Accumulator
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual intrusive_ptr<const Value> getValue() const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual Value getValue() const;
         virtual const char *getOpName() const;
 
         /*
diff --git a/src/mongo/db/pipeline/accumulator_add_to_set.cpp b/src/mongo/db/pipeline/accumulator_add_to_set.cpp
index 86d4366ff0c..81c96ff9091 100755
--- a/src/mongo/db/pipeline/accumulator_add_to_set.cpp
+++ b/src/mongo/db/pipeline/accumulator_add_to_set.cpp
@@ -21,12 +21,11 @@
 #include "db/pipeline/value.h"
 
 namespace mongo {
-    intrusive_ptr<const Value> AccumulatorAddToSet::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value AccumulatorAddToSet::evaluate(const Document& pDocument) const {
         verify(vpOperand.size() == 1);
-        intrusive_ptr<const Value> prhs(vpOperand[0]->evaluate(pDocument));
+        Value prhs(vpOperand[0]->evaluate(pDocument));
 
-        if (prhs->getType() == Undefined)
+        if (prhs.getType() == Undefined)
             ; /* nothing to add to the array */
         else if (!pCtx->getDoingMerge())
             set.insert(prhs);
@@ -37,20 +36,17 @@ namespace mongo {
               If we didn't, then we'd get an array of arrays, with one array
               from each shard that responds.
              */
-            verify(prhs->getType() == Array);
+            verify(prhs.getType() == Array);
             
-            intrusive_ptr<ValueIterator> pvi(prhs->getArray());
-            while(pvi->more()) {
-                intrusive_ptr<const Value> pElement(pvi->next());
-                set.insert(pElement);
-            }
+            const vector<Value>& array = prhs.getArray();
+            set.insert(array.begin(), array.end());
         }
 
-        return Value::getNull();
+        return Value();
     }
 
-    intrusive_ptr<const Value> AccumulatorAddToSet::getValue() const {
-        vector<intrusive_ptr<const Value> > valVec;
+    Value AccumulatorAddToSet::getValue() const {
+        vector<Value> valVec;
 
         for (itr = set.begin(); itr != set.end(); ++itr) {
             valVec.push_back(*itr);
diff --git a/src/mongo/db/pipeline/accumulator_avg.cpp b/src/mongo/db/pipeline/accumulator_avg.cpp
index 3e69b204fbb..07787325992 100755
--- a/src/mongo/db/pipeline/accumulator_avg.cpp
+++ b/src/mongo/db/pipeline/accumulator_avg.cpp
@@ -26,8 +26,7 @@ namespace mongo {
     const char AccumulatorAvg::subTotalName[] = "subTotal";
     const char AccumulatorAvg::countName[] = "count";
 
-    intrusive_ptr<const Value> AccumulatorAvg::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value AccumulatorAvg::evaluate(const Document& pDocument) const {
         if (!pCtx->getDoingMerge()) {
             Super::evaluate(pDocument);
         }
@@ -37,20 +36,19 @@ namespace mongo {
               both a subtotal and a count.  This is what getValue() produced
               below.
              */
-            intrusive_ptr<const Value> prhs(vpOperand[0]->evaluate(pDocument));
-            verify(prhs->getType() == Object);
-            intrusive_ptr<Document> pShardDoc(prhs->getDocument());
+            Value shardOut = vpOperand[0]->evaluate(pDocument);
+            verify(shardOut.getType() == Object);
 
-            intrusive_ptr<const Value> pSubTotal(pShardDoc->getValue(subTotalName));
-            verify(pSubTotal.get());
-            doubleTotal += pSubTotal->getDouble();
+            Value subTotal = shardOut[subTotalName];
+            verify(!subTotal.missing());
+            doubleTotal += subTotal.getDouble();
                 
-            intrusive_ptr<const Value> pCount(pShardDoc->getValue(countName));
-            verify(pCount.get());
-            count += pCount->getLong();
+            Value subCount = shardOut[countName];
+            verify(!subCount.missing());
+            count += subCount.getLong();
         }
 
-        return Value::getZero();
+        return Value();
     }
 
     intrusive_ptr<Accumulator> AccumulatorAvg::create(
@@ -59,7 +57,7 @@ namespace mongo {
         return pA;
     }
 
-    intrusive_ptr<const Value> AccumulatorAvg::getValue() const {
+    Value AccumulatorAvg::getValue() const {
         if (!pCtx->getInShard()) {
             double avg = 0;
             if (count)
@@ -68,11 +66,11 @@ namespace mongo {
             return Value::createDouble(avg);
         }
 
-        intrusive_ptr<Document> pDocument(Document::create());
-        pDocument->addField(subTotalName, Value::createDouble(doubleTotal));
-        pDocument->addField(countName, Value::createLong(count));
+        MutableDocument out;
+        out.addField(subTotalName, Value::createDouble(doubleTotal));
+        out.addField(countName, Value::createLong(count));
 
-        return Value::createDocument(pDocument);
+        return Value::createDocument(out.freeze());
     }
 
     AccumulatorAvg::AccumulatorAvg(
diff --git a/src/mongo/db/pipeline/accumulator_first.cpp b/src/mongo/db/pipeline/accumulator_first.cpp
index 53d8f9595e9..b226fb552d6 100755
--- a/src/mongo/db/pipeline/accumulator_first.cpp
+++ b/src/mongo/db/pipeline/accumulator_first.cpp
@@ -21,12 +21,11 @@
 
 namespace mongo {
 
-    intrusive_ptr<const Value> AccumulatorFirst::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value AccumulatorFirst::evaluate(const Document& pDocument) const {
         verify(vpOperand.size() == 1);
 
         /* only remember the first value seen */
-        if (!pValue.get())
+        if (pValue.missing())
             pValue = vpOperand[0]->evaluate(pDocument);
 
         return pValue;
diff --git a/src/mongo/db/pipeline/accumulator_last.cpp b/src/mongo/db/pipeline/accumulator_last.cpp
index d934e64111b..ea65094d04b 100755
--- a/src/mongo/db/pipeline/accumulator_last.cpp
+++ b/src/mongo/db/pipeline/accumulator_last.cpp
@@ -21,8 +21,7 @@
 
 namespace mongo {
 
-    intrusive_ptr<const Value> AccumulatorLast::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value AccumulatorLast::evaluate(const Document& pDocument) const {
         verify(vpOperand.size() == 1);
 
         /* always remember the last value seen */
diff --git a/src/mongo/db/pipeline/accumulator_min_max.cpp b/src/mongo/db/pipeline/accumulator_min_max.cpp
index aec461bab02..bc48b6fb480 100755
--- a/src/mongo/db/pipeline/accumulator_min_max.cpp
+++ b/src/mongo/db/pipeline/accumulator_min_max.cpp
@@ -21,13 +21,12 @@
 
 namespace mongo {
 
-    intrusive_ptr<const Value> AccumulatorMinMax::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value AccumulatorMinMax::evaluate(const Document& pDocument) const {
         verify(vpOperand.size() == 1);
-        intrusive_ptr<const Value> prhs(vpOperand[0]->evaluate(pDocument));
+        Value prhs(vpOperand[0]->evaluate(pDocument));
 
         /* if this is the first value, just use it */
-        if (!pValue.get())
+        if (pValue.missing())
             pValue = prhs;
         else {
             /* compare with the current value; swap if appropriate */
diff --git a/src/mongo/db/pipeline/accumulator_push.cpp b/src/mongo/db/pipeline/accumulator_push.cpp
index b7a6370d77f..4530b798465 100755
--- a/src/mongo/db/pipeline/accumulator_push.cpp
+++ b/src/mongo/db/pipeline/accumulator_push.cpp
@@ -21,15 +21,16 @@
 #include "db/pipeline/value.h"
 
 namespace mongo {
-    intrusive_ptr<const Value> AccumulatorPush::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value AccumulatorPush::evaluate(const Document& pDocument) const {
         verify(vpOperand.size() == 1);
-        intrusive_ptr<const Value> prhs(vpOperand[0]->evaluate(pDocument));
+        Value prhs(vpOperand[0]->evaluate(pDocument));
 
-        if (prhs->getType() == Undefined)
-            ; /* nothing to add to the array */
-        else if (!pCtx->getDoingMerge())
+        if (prhs.getType() == Undefined) {
+            /* nothing to add to the array */
+        }
+        else if (!pCtx->getDoingMerge()) {
             vpValue.push_back(prhs);
+        }
         else {
             /*
               If we're in the router, we need to take apart the arrays we
@@ -37,19 +38,16 @@ namespace mongo {
               If we didn't, then we'd get an array of arrays, with one array
               from each shard that responds.
              */
-            verify(prhs->getType() == Array);
+            verify(prhs.getType() == Array);
             
-            intrusive_ptr<ValueIterator> pvi(prhs->getArray());
-            while(pvi->more()) {
-                intrusive_ptr<const Value> pElement(pvi->next());
-                vpValue.push_back(pElement);
-            }
+            const vector<Value>& vec = prhs.getArray();
+            vpValue.insert(vpValue.end(), vec.begin(), vec.end());
         }
 
-        return Value::getNull();
+        return Value();
     }
 
-    intrusive_ptr<const Value> AccumulatorPush::getValue() const {
+    Value AccumulatorPush::getValue() const {
         return Value::createArray(vpValue);
     }
 
diff --git a/src/mongo/db/pipeline/accumulator_single_value.cpp b/src/mongo/db/pipeline/accumulator_single_value.cpp
index ea12ee333a2..76187d52222 100755
--- a/src/mongo/db/pipeline/accumulator_single_value.cpp
+++ b/src/mongo/db/pipeline/accumulator_single_value.cpp
@@ -21,12 +21,12 @@
 
 namespace mongo {
 
-    intrusive_ptr<const Value> AccumulatorSingleValue::getValue() const {
+    Value AccumulatorSingleValue::getValue() const {
         return pValue;
     }
 
     AccumulatorSingleValue::AccumulatorSingleValue():
-        pValue(intrusive_ptr<const Value>()) {
+        pValue(Value()) {
     }
 
 }
diff --git a/src/mongo/db/pipeline/accumulator_sum.cpp b/src/mongo/db/pipeline/accumulator_sum.cpp
index 775158c72cf..6fe091f3cf4 100755
--- a/src/mongo/db/pipeline/accumulator_sum.cpp
+++ b/src/mongo/db/pipeline/accumulator_sum.cpp
@@ -21,27 +21,26 @@
 
 namespace mongo {
 
-    intrusive_ptr<const Value> AccumulatorSum::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value AccumulatorSum::evaluate(const Document& pDocument) const {
         verify(vpOperand.size() == 1);
-        intrusive_ptr<const Value> prhs(vpOperand[0]->evaluate(pDocument));
+        Value prhs(vpOperand[0]->evaluate(pDocument));
 
-        BSONType rhsType = prhs->getType();
+        BSONType rhsType = prhs.getType();
 
         // do nothing with non numeric types
         if (!(rhsType == NumberInt || rhsType == NumberLong || rhsType == NumberDouble))
-            return Value::getZero();
+            return Value(0);
 
         // upgrade to the widest type required to hold the result
         totalType = Value::getWidestNumeric(totalType, rhsType);
 
         if (totalType == NumberInt || totalType == NumberLong) {
-            long long v = prhs->coerceToLong();
+            long long v = prhs.coerceToLong();
             longTotal += v;
             doubleTotal += v;
         }
         else if (totalType == NumberDouble) {
-            double v = prhs->coerceToDouble();
+            double v = prhs.coerceToDouble();
             doubleTotal += v;
         }
         else {
@@ -51,7 +50,7 @@ namespace mongo {
 
         count++;
 
-        return Value::getZero();
+        return Value();
     }
 
     intrusive_ptr<Accumulator> AccumulatorSum::create(
@@ -60,7 +59,7 @@ namespace mongo {
         return pSummer;
     }
 
-    intrusive_ptr<const Value> AccumulatorSum::getValue() const {
+    Value AccumulatorSum::getValue() const {
         if (totalType == NumberLong) {
             return Value::createLong(longTotal);
         }
diff --git a/src/mongo/db/pipeline/builder.cpp b/src/mongo/db/pipeline/builder.cpp
index a85d75ce02f..69eddcf9747 100755
--- a/src/mongo/db/pipeline/builder.cpp
+++ b/src/mongo/db/pipeline/builder.cpp
@@ -46,7 +46,7 @@ namespace mongo {
         pBuilder->append(fieldName, d);
     }
 
-    void BuilderObj::append(const std::string& s) {
+    void BuilderObj::append(StringData s) {
         pBuilder->append(fieldName, s);
     }
 
@@ -70,8 +70,7 @@ namespace mongo {
         pBuilder->appendTimestamp(fieldName, ot.getSecs(), ot.getInc());
     }
 
-    BuilderObj::BuilderObj(
-        BSONObjBuilder *pObjBuilder, const std::string& theFieldName):
+    BuilderObj::BuilderObj(BSONObjBuilder *pObjBuilder, StringData theFieldName):
         pBuilder(pObjBuilder),
         fieldName(theFieldName) {
     }
@@ -101,7 +100,7 @@ namespace mongo {
         pBuilder->append(d);
     }
 
-    void BuilderArray::append(const std::string& s) {
+    void BuilderArray::append(StringData s) {
         pBuilder->append(s);
     }
 
diff --git a/src/mongo/db/pipeline/builder.h b/src/mongo/db/pipeline/builder.h
index e6a68a72a32..5e4f5e48e38 100755
--- a/src/mongo/db/pipeline/builder.h
+++ b/src/mongo/db/pipeline/builder.h
@@ -44,7 +44,7 @@ namespace mongo {
         virtual void append(int i) = 0;
         virtual void append(long long ll) = 0;
         virtual void append(double d) = 0;
-        virtual void append(const std::string& s) = 0;
+        virtual void append(StringData s) = 0;
         virtual void append(const OID &o) = 0;
         virtual void append(const Date_t &d) = 0;
         virtual void append(const OpTime& ot) = 0;
@@ -62,18 +62,18 @@ namespace mongo {
         virtual void append(int i);
         virtual void append(long long ll);
         virtual void append(double d);
-        virtual void append(const std::string& s);
+        virtual void append(StringData s);
         virtual void append(const OID &o);
         virtual void append(const Date_t &d);
         virtual void append(const OpTime& ot);
         virtual void append(BSONObjBuilder *pDone);
         virtual void append(BSONArrayBuilder *pDone);
 
-        BuilderObj(BSONObjBuilder *pBuilder, const std::string& fieldName);
+        BuilderObj(BSONObjBuilder *pBuilder, StringData fieldName);
 
     private:
         BSONObjBuilder *pBuilder;
-        string fieldName;
+        StringData fieldName;
     };
 
     class BuilderArray :
@@ -86,7 +86,7 @@ namespace mongo {
         virtual void append(int i);
         virtual void append(long long ll);
         virtual void append(double d);
-        virtual void append(const std::string& s);
+        virtual void append(StringData s);
         virtual void append(const OID &o);
         virtual void append(const Date_t &d);
         virtual void append(const OpTime& ot);
diff --git a/src/mongo/db/pipeline/document.cpp b/src/mongo/db/pipeline/document.cpp
index e34af40b8dd..65b0ec6e2de 100755..100644
--- a/src/mongo/db/pipeline/document.cpp
+++ b/src/mongo/db/pipeline/document.cpp
@@ -15,159 +15,292 @@
  */
 
 #include "pch.h"
+
+#include "mongo/db/pipeline/document.h"
+
 #include <boost/functional/hash.hpp>
-#include "db/jsobj.h"
-#include "db/pipeline/document.h"
-#include "db/pipeline/value.h"
-#include "util/mongoutils/str.h"
+
+#include "mongo/db/jsobj.h"
+#include "mongo/db/pipeline/field_path.h"
+#include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
     using namespace mongoutils;
 
-    string Document::idName("_id");
+    Position DocumentStorage::findField(StringData requested) const {
+        if (_numFields >= HASH_TAB_MIN) { // hash lookup
+            const unsigned bucket = bucketForKey(requested);
 
-    intrusive_ptr<Document> Document::createFromBsonObj(BSONObj* pBsonObj) {
-        return new Document(pBsonObj);
-    }
+            Position pos = _hashTab[bucket];
+            while (pos.found()) {
+                const ValueElement& elem = getField(pos);
+                if (str::equals(requested.data(), elem.name))
+                    return pos;
 
-    Document::Document(BSONObj* pBsonObj) {
-        const int fields = pBsonObj->nFields();
-        vFieldName.reserve(fields);
-        vpValue.reserve(fields);
-        BSONObjIterator bsonIterator(pBsonObj->begin());
-        while(bsonIterator.more()) {
-            BSONElement bsonElement(bsonIterator.next());
-            string fieldName(bsonElement.fieldName());
+                // possible collision
+                pos = elem.nextCollision;
+            }
+        }
+        else if (_numFields) { // linear scan
+            for (DocumentStorageIterator it = iteratorAll(); !it.atEnd(); it.advance()) {
+                if (size_t(it->nameLen) == requested.size()
+                    && str::equals(it->name, requested.data())) {
+                    return it.position();
+                }
+            }
+        }
 
-            // LATER grovel through structures???
-            intrusive_ptr<const Value> pValue(
-                Value::createFromBsonElement(&bsonElement));
+        // if we got here, there's no such field
+        return Position();
+    }
 
-            vFieldName.push_back(fieldName);
-            vpValue.push_back(pValue);
+    Value& DocumentStorage::appendField(StringData name) {
+        Position pos = getNextPosition();
+        const int nameSize = name.size();
+
+        // these are the same for everyone
+        const Position nextCollision;
+        const Value value;
+
+        // Make room for new field (and padding at end for alignment)
+        const unsigned newUsed = ValueElement::align(_usedBytes + sizeof(ValueElement) + nameSize);
+        if (_buffer + newUsed > _bufferEnd)
+            alloc(newUsed);
+        _usedBytes = newUsed;
+
+        // Append structure of a ValueElement
+        char* dest = _buffer + pos.index; // must be after alloc since it changes _buffer
+#define append(x) memcpy(dest, &(x), sizeof(x)); dest += sizeof(x)
+        append(value);
+        append(nextCollision);
+        append(nameSize);
+        memcpy(dest, name.data(), nameSize + 1/*NUL*/);
+        // Padding for alignment handled above
+#undef append
+
+        // Make sure next field starts where we expect it
+        fassert(16486, getField(pos).next()->ptr() == _buffer + _usedBytes);
+
+        _numFields++;
+
+        if (_numFields > HASH_TAB_MIN) {
+            addFieldToHashTable(pos);
+        }
+        else if (_numFields == HASH_TAB_MIN) {
+            // adds all fields to hash table (including the one we just added)
+            rehash();
         }
-    }
 
-    void Document::toBson(BSONObjBuilder* pBuilder) const {
-        const size_t n = vFieldName.size();
-        for(size_t i = 0; i < n; ++i)
-            vpValue[i]->addToBsonObj(pBuilder, vFieldName[i]);
+        return getField(pos);
     }
 
-    intrusive_ptr<Document> Document::create(size_t sizeHint) {
-        intrusive_ptr<Document> pDocument(new Document(sizeHint));
-        return pDocument;
+    // Call after adding field to _fields and increasing _numFields
+    void DocumentStorage::addFieldToHashTable(Position pos) {
+        ValueElement& elem = getField(pos);
+        elem.nextCollision = Position();
+
+        const unsigned bucket = bucketForKey(elem.nameSD());
+
+        Position* posPtr = &_hashTab[bucket];
+        while (posPtr->found()) {
+            // collision: walk links and add new to end
+            posPtr = &getField(*posPtr).nextCollision;
+        }
+        *posPtr = Position(pos.index);
     }
 
-    Document::Document(size_t sizeHint):
-        vFieldName(),
-        vpValue() {
-        if (sizeHint) {
-            vFieldName.reserve(sizeHint);
-            vpValue.reserve(sizeHint);
+    void DocumentStorage::alloc(unsigned newSize) {
+        const bool firstAlloc = !_buffer;
+        const bool doingRehash = needRehash();
+        const size_t oldCapacity = _bufferEnd - _buffer;
+
+        // make new bucket count big enough
+        while (needRehash() || hashTabBuckets() < HASH_TAB_INIT_SIZE)
+            _hashTabMask = hashTabBuckets()*2 - 1;
+
+        // only allocate power-of-two sized space > 128 bytes
+        size_t capacity = 128;
+        while (capacity < newSize + hashTabBytes())
+            capacity *= 2;
+
+        uassert(16490, "Tried to make oversized document",
+                capacity <= size_t(BufferMaxSize));
+
+        boost::scoped_array<char> oldBuf(_buffer);
+        _buffer = new char[capacity];
+        _bufferEnd = _buffer + capacity - hashTabBytes();
+
+        if (!firstAlloc) {
+            // This just copies the elements
+            memcpy(_buffer, oldBuf.get(), _usedBytes);
+
+            if (_numFields >= HASH_TAB_MIN) {
+                // if we were hashing, deal with the hash table
+                if (doingRehash) {
+                    rehash();
+                }
+                else {
+                    // no rehash needed so just slide table down to new position
+                    memcpy(_hashTab, oldBuf.get() + oldCapacity, hashTabBytes());
+                }
+            }
         }
     }
 
-    intrusive_ptr<Document> Document::clone() {
-        const size_t n = vFieldName.size();
-        intrusive_ptr<Document> pNew(Document::create(n));
-        for(size_t i = 0; i < n; ++i)
-            pNew->addField(vFieldName[i], vpValue[i]);
+    void DocumentStorage::reserveFields(size_t expectedFields) {
+        fassert(16487, !_buffer);
+
+        unsigned buckets = HASH_TAB_INIT_SIZE;
+        while (buckets < expectedFields)
+            buckets *= 2;
+        _hashTabMask = buckets - 1;
+
+        // Using expectedFields+1 to allow space for long field names
+        const size_t newSize = (expectedFields+1) * ValueElement::align(sizeof(ValueElement));
 
-        return pNew;
+        uassert(16491, "Tried to make oversized document",
+                newSize <= size_t(BufferMaxSize));
+
+        _buffer = new char[newSize + hashTabBytes()];
+        _bufferEnd = _buffer + newSize;
     }
 
-    Document::~Document() {
+    intrusive_ptr<DocumentStorage> DocumentStorage::clone() const {
+        intrusive_ptr<DocumentStorage> out (new DocumentStorage());
+
+        // Make a copy of the buffer.
+        // It is very important that the positions of each field are the same after cloning.
+        const size_t bufferBytes = (_bufferEnd + hashTabBytes()) - _buffer;
+        out->_buffer = new char[bufferBytes];
+        out->_bufferEnd = out->_buffer + (_bufferEnd - _buffer);
+        memcpy(out->_buffer, _buffer, bufferBytes);
+
+        // Copy remaining fields
+        out->_usedBytes = _usedBytes;
+        out->_numFields = _numFields;
+        out->_hashTabMask = _hashTabMask;
+
+        // Tell values that they have been memcpyed (updates ref counts)
+        for (DocumentStorageIterator it = out->iteratorAll(); !it.atEnd(); it.advance()) {
+            it->val.memcpyed();
+        }
+
+        return out;
     }
 
-    FieldIterator *Document::createFieldIterator() {
-        return new FieldIterator(intrusive_ptr<Document>(this));
+    DocumentStorage::~DocumentStorage() {
+        boost::scoped_array<char> deleteBufferAtScopeEnd (_buffer);
+
+        for (DocumentStorageIterator it = iteratorAll(); !it.atEnd(); it.advance()) {
+            it->val.~Value(); // explicit destructor call
+        }
     }
 
-    intrusive_ptr<const Value> Document::getValue(const string &fieldName) {
-        /*
-          For now, assume the number of fields is small enough that iteration
-          is ok.  Later, if this gets large, we can create a map into the
-          vector for these lookups.
+    Document::Document(const BSONObj& bson) {
+        MutableDocument md(bson.nFields());
 
-          Note that because of the schema-less nature of this data, we always
-          have to look, and can't assume that the requested field is always
-          in a particular place as we would with a statically compilable
-          reference.
-        */
-        const size_t n = vFieldName.size();
-        for(size_t i = 0; i < n; ++i) {
-            if (fieldName.compare(vFieldName[i]) == 0)
-                return vpValue[i];
+        BSONObjIterator it(bson);
+        while(it.more()) {
+            BSONElement bsonElement(it.next());
+            md.addField(bsonElement.fieldName(), Value(bsonElement));
         }
 
-        return(intrusive_ptr<const Value>());
+        *this = md.freeze();
     }
 
-    void Document::addField(const string &fieldName,
-                            const intrusive_ptr<const Value> &pValue) {
-        vFieldName.push_back(fieldName);
-        vpValue.push_back(pValue);
+    void Document::toBson(BSONObjBuilder* pBuilder) const {
+        for (DocumentStorageIterator it = storage().iterator(); !it.atEnd(); it.advance()) {
+            it->val.addToBsonObj(pBuilder, it->nameSD());
+        }
     }
 
-    void Document::setField(size_t index,
-                            const string &fieldName,
-                            const intrusive_ptr<const Value> &pValue) {
-        /* special case:  should this field be removed? */
-        if (!pValue.get()) {
-            vFieldName.erase(vFieldName.begin() + index);
-            vpValue.erase(vpValue.begin() + index);
-            return;
+    MutableDocument::MutableDocument(size_t expectedFields)
+        : _storageHolder(NULL)
+        , _storage(_storageHolder)
+    {
+        if (expectedFields) {
+            storage().reserveFields(expectedFields);
         }
-
-        /* set the indicated field */
-        vFieldName[index] = fieldName;
-        vpValue[index] = pValue;
     }
 
-    intrusive_ptr<const Value> Document::getField(const string &fieldName) const {
-        const size_t n = vFieldName.size();
-        for(size_t i = 0; i < n; ++i) {
-            if (fieldName.compare(vFieldName[i]) == 0)
-                return vpValue[i];
+    MutableValue MutableDocument::getNestedFieldHelper(MutableDocument& md,
+                                                       const vector<Position>& positions,
+                                                       size_t level) {
+        fassert(16488, !positions.empty());
+
+        if (level == positions.size()-1) {
+            return md[positions[level]];
         }
+        else {
+            MutableDocument nested (md[positions[level]]);
+            return getNestedFieldHelper(nested, positions, level+1);
+        }
+    }
 
-        /* if we got here, there's no such field */
-        return intrusive_ptr<const Value>();
+    MutableValue MutableDocument::getNestedField(const vector<Position>& positions) {
+        return getNestedFieldHelper(*this, positions, 0);
     }
 
-    size_t Document::getApproximateSize() const {
-        size_t size = sizeof(Document);
-        const size_t n = vpValue.size();
-        for(size_t i = 0; i < n; ++i)
-            size += vpValue[i]->getApproximateSize();
+    static Value getNestedFieldHelper(const Document& doc,
+                                      const FieldPath& fieldNames,
+                                      vector<Position>* positions,
+                                      size_t level) {
 
-        return size;
+        fassert(16489, fieldNames.getPathLength());
+
+        const string& fieldName = fieldNames.getFieldName(level);
+        const Position pos = doc.positionOf(fieldName);
+
+        if (!pos.found())
+            return Value();
+
+        if (positions)
+            positions->push_back(pos);
+
+        if (level == fieldNames.getPathLength()-1)
+            return doc.getField(pos);
+
+        Value val = doc.getField(pos);
+        if (val.getType() != Object)
+            return Value();
+
+        return getNestedFieldHelper(val.getDocument(), fieldNames, positions, level+1);
+    }
+
+    const Value Document::getNestedField(const FieldPath& fieldNames,
+                                         vector<Position>* positions) const {
+        return getNestedFieldHelper(*this, fieldNames, positions, 0);
     }
 
-    size_t Document::getFieldIndex(const string &fieldName) const {
-        const size_t n = vFieldName.size();
-        size_t i = 0;
-        for(; i < n; ++i) {
-            if (fieldName.compare(vFieldName[i]) == 0)
-                break;
+    size_t Document::getApproximateSize() const {
+        if (!_storage)
+            return 0; // we've allocated no memory
+
+        size_t size = sizeof(DocumentStorage);
+        size += storage().allocatedBytes();
+
+        for (DocumentStorageIterator it = storage().iterator(); !it.atEnd(); it.advance()) {
+            size += it->val.getApproximateSize();
+            size -= sizeof(Value); // already accounted for above
         }
 
-        return i;
+        return size;
     }
 
     void Document::hash_combine(size_t &seed) const {
-        const size_t n = vFieldName.size();
-        for(size_t i = 0; i < n; ++i) {
-            boost::hash_combine(seed, vFieldName[i]);
-            vpValue[i]->hash_combine(seed);
+        for (DocumentStorageIterator it = storage().iterator(); !it.atEnd(); it.advance()) {
+            StringData name = it->nameSD();
+            boost::hash_range(seed, name.data(), name.data()+name.size());
+            it->val.hash_combine(seed);
         }
     }
 
-    int Document::compare(const intrusive_ptr<Document> &rL,
-                          const intrusive_ptr<Document> &rR) {
-        const size_t lSize = rL->vFieldName.size();
-        const size_t rSize = rR->vFieldName.size();
+    int Document::compare(const Document& rL, const Document& rR) {
+        const size_t lSize = rL->storage().size();
+        const size_t rSize = rR->storage().size();
+
+        DocumentStorageIterator lIt = rL.storage().iterator();
+        DocumentStorageIterator rIt = rR.storage().iterator();
 
         for(size_t i = 0; true; ++i) {
             if (i >= lSize) {
@@ -180,43 +313,35 @@ namespace mongo {
             if (i >= rSize)
                 return 1; // right document is shorter
 
-            const int nameCmp = rL->vFieldName[i].compare(rR->vFieldName[i]);
+            const ValueElement& rField = rIt.get();
+            const ValueElement& lField = lIt.get();
+
+            rIt.advance();
+            lIt.advance();
+
+            const int nameCmp = strcmp(lField.name, rField.name);
             if (nameCmp)
                 return nameCmp; // field names are unequal
 
-            const int valueCmp = Value::compare(rL->vpValue[i], rR->vpValue[i]);
+            const int valueCmp = Value::compare(lField.val, rField.val);
             if (valueCmp)
                 return valueCmp; // fields are unequal
         }
-
-        /* NOTREACHED */
-        verify(false);
-        return 0;
     }
 
     string Document::toString() const {
-        // this is a temporary hack and it should only be used for debugging
-        BSONObjBuilder bb;
-        toBson(&bb);
-        return bb.done().toString();
-    }
-
-    /* ----------------------- FieldIterator ------------------------------- */
+        if (empty())
+            return "{}";
 
-    FieldIterator::FieldIterator(const intrusive_ptr<Document> &pTheDocument):
-        pDocument(pTheDocument),
-        index(0) {
-    }
+        StringBuilder out;
+        const char* prefix = "{";
 
-    bool FieldIterator::more() const {
-        return (index < pDocument->vFieldName.size());
-    }
+        for (DocumentStorageIterator it = storage().iterator(); !it.atEnd(); it.advance()) {
+            out << prefix << it->name << ": " << it->val.toString();
+            prefix = ", ";
+        }
+        out << '}';
 
-    pair<string, intrusive_ptr<const Value> > FieldIterator::next() {
-        verify(more());
-        pair<string, intrusive_ptr<const Value> > result(
-            pDocument->vFieldName[index], pDocument->vpValue[index]);
-        ++index;
-        return result;
+        return out.str();
     }
 }
diff --git a/src/mongo/db/pipeline/document.h b/src/mongo/db/pipeline/document.h
index 3db7e9dfebf..09b1a356011 100755..100644
--- a/src/mongo/db/pipeline/document.h
+++ b/src/mongo/db/pipeline/document.h
@@ -16,234 +16,389 @@
 
 #pragma once
 
-#include "mongo/pch.h"
+#include "mongo/db/pipeline/document_internal.h"
 
-#include "util/intrusive_counter.h"
+#include <boost/functional/hash.hpp>
+#include "mongo/bson/util/builder.h"
 
 namespace mongo {
     class BSONObj;
     class FieldIterator;
+    class FieldPath;
     class Value;
-
-    class Document :
-        public IntrusiveCounterUnsigned {
+    class MutableDocument;
+
+    /** An internal class that represents the position of a field in a document.
+     *
+     *  This is a low-level class that you usually don't need to worry about.
+     *
+     *  The main use of this class for clients is to allow refetching or
+     *  setting a field without looking it up again. It has a default
+     *  constructor that represents a field not being in a document. It also
+     *  has a method 'bool found()' that tells you if a field was found.
+     *
+     *  For more details see document_internal.h
+     */
+    class Position;
+
+    /** A Document is similar to a BSONObj but with a different in-memory representation.
+     *
+     *  A Document can be treated as a const map<string, const Value> that is
+     *  very cheap to copy and is Assignable.  Therefore, it is acceptable to
+     *  pass and return by Value. Note that the data in a Document is
+     *  immutable, but you can replace a Document instance with assignment.
+     *
+     *  See Also: Value class in Value.h
+     */
+    class Document {
     public:
-        ~Document();
 
-        /*
-          Create a new Document from the given BSONObj.
+        /// Empty Document (does no allocation)
+        Document() {}
 
-          Document field values may be pointed to in the BSONObj, so it
-          must live at least as long as the resulting Document.
+        /// Create a new Document deep-converted from the given BSONObj.
+        explicit Document(const BSONObj& bson);
 
-          @returns shared pointer to the newly created Document
-        */
-        static intrusive_ptr<Document> createFromBsonObj(BSONObj* pBsonObj);
+        void swap(Document& rhs) { _storage.swap(rhs._storage); }
 
-        /*
-          Create a new empty Document.
+        /// Look up a field by key name. Returns Value() if no such field. O(1)
+        const Value operator[] (StringData key) const { return getField(key); }
+        const Value getField(StringData key) const { return storage().getField(key); }
 
-          @param sizeHint a hint at what the number of fields will be; if
-            known, this can be used to increase memory allocation efficiency
-          @returns shared pointer to the newly created Document
-        */
-        static intrusive_ptr<Document> create(size_t sizeHint = 0);
+        /// Look up a field by Position. See positionOf and getNestedField.
+        const Value operator[] (Position pos) const { return getField(pos); }
+        const Value getField(Position pos) const { return storage().getField(pos); }
 
-        /*
-          Clone a document.
+        /** Similar to BSONObj::getFieldDotted, but using FieldPath rather than a dotted string.
+         *  If you pass a non-NULL positions vector, you get back a path suitable
+         *  to pass to MutableDocument::setNestedField.
+         *
+         *  TODO a version that doesn't use FieldPath
+         */
+        const Value getNestedField(const FieldPath& fieldNames,
+                                   vector<Position>* positions=NULL) const;
 
-          The new document shares all the fields' values with the original.
+        /// Number of fields in this document. O(n)
+        size_t size() const { return storage().size(); }
 
-          This is not a deep copy.  Only the fields on the top-level document
-          are cloned.
+        /// True if this document has no fields.
+        bool empty() const { return !_storage || storage().iterator().atEnd(); }
 
-          @returns the shallow clone of the document
-        */
-        intrusive_ptr<Document> clone();
+        /// Create a new FieldIterator that can be used to examine the Document's fields in order.
+        FieldIterator fieldIterator() const;
 
-        /*
-          Add this document to the BSONObj under construction with the
-          given BSONObjBuilder.
-        */
-        void toBson(BSONObjBuilder *pBsonObjBuilder) const;
+        /// Convenience type for dealing with fields. Used by FieldIterator.
+        typedef pair<StringData, Value> FieldPair;
 
-        /*
-          Create a new FieldIterator that can be used to examine the
-          Document's fields.
-        */
-        FieldIterator *createFieldIterator();
-
-        /*
-          Get the value of the specified field.
-
-          @param fieldName the name of the field
-          @return point to the requested field
-        */
-        intrusive_ptr<const Value> getValue(const string &fieldName);
-
-        /*
-          Add the given field to the Document.
-
-          BSON documents' fields are ordered; the new Field will be
-          appended to the current list of fields.
-
-          It is an error to add a field that has the same name as another
-          field.
-        */
-        void addField(const string &fieldName,
-                      const intrusive_ptr<const Value> &pValue);
-
-        /*
-          Set the given field to be at the specified position in the
-          Document.  This will replace any field that is currently in that
-          position.  The index must be within the current range of field
-          indices, otherwise behavior is undefined.
-
-          pValue.get() may be NULL, in which case the field will be
-          removed.  fieldName is ignored in this case.
-
-          @param index the field index in the list of fields
-          @param fieldName the new field name
-          @param pValue the new Value
-        */
-        void setField(size_t index,
-                      const string &fieldName,
-                      const intrusive_ptr<const Value> &pValue);
-
-        /*
-          Convenience type for dealing with fields.
+        /** Get the approximate storage size of the document and sub-values in bytes.
+         *  Note: Some memory may be shared with other Documents or between fields within
+         *        a single Document so this can overestimate usage.
          */
-        typedef pair<string, intrusive_ptr<const Value> > FieldPair;
-
-        /*
-          Get the indicated field.
+        size_t getApproximateSize() const;
 
-          @param index the field index in the list of fields
-          @returns the field name and value of the field
+        /** Compare two documents.
+         *
+         *  BSON document field order is significant, so this just goes through
+         *  the fields in order.  The comparison is done in roughly the same way
+         *  as strings are compared, but comparing one field at a time instead
+         *  of one character at a time.
+         *
+         *  @returns an integer less than zero, zero, or an integer greater than
+         *           zero, depending on whether lhs < rhs, lhs == rhs, or lhs > rhs
+         *  Warning: may return values other than -1, 0, or 1
          */
-        FieldPair getField(size_t index) const;
+        static int compare(const Document& lhs, const Document& rhs);
 
-        /*
-          Get the number of fields in the Document.
+        string toString() const; // TODO support streams
 
-          @returns the number of fields in the Document
+        /** Calculate a hash value.
+         *
+         * Meant to be used to create composite hashes suitable for
+         * hashed container classes such as unordered_map.
          */
-        size_t getFieldCount() const;
+        void hash_combine(size_t &seed) const;
 
-        /*
-          Get the index of the given field.
+        /// Add this document to the BSONObj under construction with the given BSONObjBuilder.
+        void toBson(BSONObjBuilder *pBsonObjBuilder) const;
 
-          @param fieldName the name of the field
-          @returns the index of the field, or if it does not exist, the number
-            of fields (getFieldCount())
-        */
-        size_t getFieldIndex(const string &fieldName) const;
+        /** Return the abstract Position of a field, suitable to pass to operator[] or getField().
+         *  This can potentially save time if you need to refer to a field multiple times.
+         */
+        Position positionOf(StringData fieldName) const { return storage().findField(fieldName); }
+
+        /** Clone a document.
+         *
+         *  This should only be called by MutableDocument and tests
+         *
+         *  The new document shares all the fields' values with the original.
+         *  This is not a deep copy.  Only the fields on the top-level document
+         *  are cloned.
+         */
+        Document clone() const { return Document(storage().clone().get()); }
 
-        /*
-          Get a field by name.
+        // TEMP for compatibility with legacy intrusive_ptr<Document>
+              Document& operator*()       { return *this; }
+        const Document& operator*() const { return *this; }
+              Document* operator->()       { return this; }
+        const Document* operator->() const { return this; }
+        const void* getPtr() const { return _storage.get(); }
+        void reset() { return _storage.reset(); }
+        static Document createFromBsonObj(BSONObj* pBsonObj) { return Document(*pBsonObj); }
+        size_t getFieldCount() const { return size(); }
+        Value getValue(StringData fieldName) const { return getField(fieldName); }
 
-          @param fieldName the name of the field
-          @returns the value of the field
-        */
-        intrusive_ptr<const Value> getField(const string &fieldName) const;
+        // TODO: replace with logical equality once all current usages are fixed
+        bool operator== (const Document& lhs) const { return _storage == lhs._storage; }
 
-        /*
-          Get the approximate storage size of the document, in bytes.
+        explicit Document(const DocumentStorage* ptr) : _storage(ptr) {};
 
-          Under the assumption that field name strings are shared, they are
-          not included in the total.
+    private:
+        friend class FieldIterator;
+        friend class ValueStorage;
+        friend class MutableDocument;
 
-          @returns the approximate storage
-        */
-        size_t getApproximateSize() const;
+        const DocumentStorage& storage() const {
+            return (_storage ? *_storage : DocumentStorage::emptyDoc());
+        }
+        intrusive_ptr<const DocumentStorage> _storage;
+    };
 
-        /*
-          Compare two documents.
+    /** This class is returned by MutableDocument to allow you to modify its values.
+     *  You are not allowed to hold variables of this type (enforced by the type system).
+     */
+    class MutableValue {
+    public:
+        void operator= (const Value& v) { _val = v; }
 
-          BSON document field order is significant, so this just goes through
-          the fields in order.  The comparison is done in roughly the same way
-          as strings are compared, but comparing one field at a time instead
-          of one character at a time.
-        */
-        static int compare(const intrusive_ptr<Document> &rL,
-                           const intrusive_ptr<Document> &rR);
+        /** These are designed to allow things like mutDoc["a"]["b"]["c"] = Value(10);
+         *  It is safe to use even on nonexistent fields.
+         */
+        MutableValue operator[] (StringData key) { return getField(key); }
+        MutableValue operator[] (Position pos)   { return getField(pos); }
 
-        static string idName; // shared "_id"
+        MutableValue getField(StringData key);
+        MutableValue getField(Position pos);
 
-        /*
-          Calculate a hash value.
+    private:
+        friend class MutableDocument;
 
-          Meant to be used to create composite hashes suitable for
-          boost classes such as unordered_map<>.
+        /// can only be constructed or copied by self and friends
+        MutableValue(const MutableValue& other): _val(other._val) {}
+        explicit MutableValue(Value& val): _val(val) {}
 
-          @param seed value to augment with this' hash
-        */
-        void hash_combine(size_t &seed) const;
+        /// Used by MutableDocument(MutableValue)
+        const RefCountable*& getDocPtr() {
+            if (_val.getType() != Object)
+                *this = Value(Document());
 
-        // For debugging purposes only!
-        string toString() const;
+            return _val._storage.genericRCPtr;
+        }
 
-    private:
-        friend class FieldIterator;
-
-        Document(size_t sizeHint);
-        Document(BSONObj* pBsonObj);
+        MutableValue& operator= (const MutableValue&); // not assignable with another MutableValue
 
-        /* these two vectors parallel each other */
-        vector<string> vFieldName;
-        vector<intrusive_ptr<const Value> > vpValue;
+        Value& _val;
     };
 
-
-    class FieldIterator :
-            boost::noncopyable {
+    /** MutableDocument is a Document builder that supports both adding and updating fields.
+     *
+     *  This class fills a similar role to BSONObjBuilder, but allows you to
+     *  change existing fields and more easily write to sub-Documents.
+     *
+     *  To preserve the immutability of Documents, MutableDocument will
+     *  shallow-clone its storage on write (COW) if it is shared with any other
+     *  Documents.
+     */
+    class MutableDocument : boost::noncopyable {
     public:
-        /*
-          Ask if there are more fields to return.
 
-          @return true if there are more fields, false otherwise
-        */
-        bool more() const;
+        /** Create a new empty Document.
+         *
+         *  @param expectedFields a hint at what the number of fields will be, if known.
+         *         this can be used to increase memory allocation efficiency. There is
+         *         no impact on correctness if this field over or under estimates.
+         *
+         *  TODO: find some way to convey field-name sizes to make even more efficient
+         */
+        MutableDocument() :_storageHolder(NULL), _storage(_storageHolder) {}
+        explicit MutableDocument(size_t expectedFields);
+
+        /// No copy yet. Copy-on-write. See storage()
+        explicit MutableDocument(const Document& d) : _storageHolder(NULL)
+                                                    , _storage(_storageHolder) {
+            reset(d);
+        }
+
+        ~MutableDocument() {
+            if (_storageHolder)
+                intrusive_ptr_release(_storageHolder);
+        }
+
+        /** Replace the current base Document with the argument
+         *
+         *  All Positions from the passed in Document are valid and refer to the
+         *  same field in this MutableDocument.
+         */
+        void reset(const Document& d=Document()) { reset(d._storage.get()); }
+
+        /** Add the given field to the Document.
+         *
+         *  BSON documents' fields are ordered; the new Field will be
+         *  appended to the current list of fields.
+         *
+         *  Unlike getField/setField, addField does not look for a field with the
+         *  same name and therefore cannot be used to update fields.
+         *
+         *  It is an error to add a field that has the same name as another field.
+         *
+         *  TODO: This is currently allowed but getField only gets first field.
+         *        Decide what level of support is needed for duplicate fields.
+         *        If duplicates are not allowed, consider removing this method.
+         */
+        void addField(StringData fieldName, const Value& val) {
+            storage().appendField(fieldName) = val;
+        }
 
-        /*
-          Move the iterator to point to the next field and return it.
+        /** Update field by key. If there is no field with that key, add one.
+         *
+         *  If the new value is missing(), the field is logically removed.
+         */
+        MutableValue operator[] (StringData key) { return getField(key); }
+        void setField(StringData key, const Value& val) { getField(key) = val; }
+        MutableValue getField(StringData key) {
+            return MutableValue(storage().getField(key));
+        }
+
+        /// Update field by Position. Must already be a valid Position.
+        MutableValue operator[] (Position pos) { return getField(pos); }
+        void setField(Position pos, const Value& val) { getField(pos) = val; }
+        MutableValue getField(Position pos) {
+            return MutableValue(storage().getField(pos).val);
+        }
+
+        /// Logically remove a field. Note that memory usage does not decrease.
+        void remove(StringData key) { getField(key) = Value(); }
+
+        /// Takes positions vector from Document::getNestedField.
+        MutableValue getNestedField(const vector<Position>& positions);
+        void setNestedField(const vector<Position>& positions, const Value& val) {
+            getNestedField(positions) = val;
+        }
+
+        /** Convert to a read-only document and release reference.
+         *
+         *  Call this to indicate that you are done with this Document and will
+         *  not be making further changes from this MutableDocument.
+         *
+         *  TODO: there are some optimizations that may make sense at freeze time.
+         */
+        Document freeze() {
+            Document ret(storagePtr());
+            reset(NULL);
+            return ret;
+        }
+
+        /** Borrow a readable reference to this Document.
+         *
+         *  Note that unlike freeze(), this indicates intention to continue
+         *  modifying this document. The returned Document will not observe
+         *  future changes to this MutableDocument.
+         */
+        Document peek() {
+            return Document(storagePtr());
+        }
 
-          @return the next field's <name, Value>
-        */
-        Document::FieldPair next();
+    private:
+        friend class MutableValue; // for access to next constructor
+        explicit MutableDocument(MutableValue mv)
+            : _storageHolder(NULL)
+            , _storage(mv.getDocPtr())
+        {}
+
+        void reset(const DocumentStorage* ds) {
+            if (_storage) intrusive_ptr_release(_storage);
+            _storage = ds;
+            if (_storage) intrusive_ptr_add_ref(_storage);
+        }
+
+        // This is split into 3 functions to speed up the fast-path
+        DocumentStorage& storage() {
+            if (MONGO_unlikely( !_storage ))
+                return newStorage();
+
+            if (MONGO_unlikely( _storage->isShared() ))
+                return clonedStorage();
+
+            // This function exists to ensure this is safe
+            return const_cast<DocumentStorage&>(*storagePtr());
+        }
+        DocumentStorage& newStorage() {
+            reset(new DocumentStorage);
+            return const_cast<DocumentStorage&>(*storagePtr());
+        }
+        DocumentStorage& clonedStorage() {
+            reset(storagePtr()->clone().get());
+            return const_cast<DocumentStorage&>(*storagePtr());
+        }
+
+        MutableValue getNestedFieldHelper(MutableDocument& md,
+                                          const vector<Position>& positions,
+                                          size_t level);
+
+        // this should only be called by storage methods and peek/freeze
+        const DocumentStorage* storagePtr() const {
+            dassert(!_storage || typeid(*_storage) == typeid(const DocumentStorage));
+            return static_cast<const DocumentStorage*>(_storage);
+        }
+
+        // These are both const to prevent modifications bypassing storage() method.
+        // They always point to NULL or an object with dynamic type DocumentStorage.
+        const RefCountable* _storageHolder; // Only used in constructors and destructor
+        const RefCountable*& _storage; // references either above member or genericRCPtr in a Value
+    };
+
+    /// This is the public iterator over a document
+    class FieldIterator {
+    public:
+        explicit FieldIterator(const Document& doc)
+            : _doc(doc)
+            , _it(_doc.storage().iterator())
+        {}
 
-        /*
-          Constructor.
+        /// Ask if there are more fields to return.
+        bool more() const { return !_it.atEnd(); }
 
-          @param pDocument points to the document whose fields are being
-              iterated
-        */
-        FieldIterator(const intrusive_ptr<Document> &pDocument);
+        /// Get next item and advance iterator
+        Document::FieldPair next() {
+            verify(more());
+
+            Document::FieldPair fp (_it->nameSD(), _it->val);
+            _it.advance();
+            return fp;
+        }
 
     private:
-        friend class Document;
-
-        /*
-          We'll hang on to the original document to ensure we keep the
-          fieldPtr vector alive.
-        */
-        intrusive_ptr<Document> pDocument;
-        size_t index; // current field in iteration
+        // We'll hang on to the original document to ensure we keep its storage alive
+        Document _doc;
+        DocumentStorageIterator _it;
     };
 }
 
+namespace std {
+    template<>
+    inline void swap(mongo::Document& lhs, mongo::Document& rhs) { lhs.swap(rhs); }
+}
 
 /* ======================= INLINED IMPLEMENTATIONS ========================== */
 
 namespace mongo {
+    inline FieldIterator Document::fieldIterator() const {
+        return FieldIterator(*this);
+    }
 
-    inline size_t Document::getFieldCount() const {
-        return vFieldName.size();
+    inline MutableValue MutableValue::getField(Position pos) {
+        return MutableDocument(*this).getField(pos);
     }
-    
-    inline Document::FieldPair Document::getField(size_t index) const {
-        verify( index < vFieldName.size() );
-        return FieldPair(vFieldName[index], vpValue[index]);
+    inline MutableValue MutableValue::getField(StringData key) {
+        return MutableDocument(*this).getField(key);
     }
-
 }
diff --git a/src/mongo/db/pipeline/document_internal.h b/src/mongo/db/pipeline/document_internal.h
new file mode 100644
index 00000000000..39677bb5698
--- /dev/null
+++ b/src/mongo/db/pipeline/document_internal.h
@@ -0,0 +1,302 @@
+/**
+ * Copyright 2012 (c) 10gen 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/>.
+ */
+
+#pragma once
+
+#include <third_party/murmurhash3/MurmurHash3.h>
+
+#include "mongo/util/intrusive_counter.h"
+#include "mongo/db/pipeline/value.h"
+
+namespace mongo {
+    /** Helper class to make the position in a document abstract
+     *  Warning: This is NOT guaranteed to be the ordered position.
+     *           eg. the first field may not be at Position(0)
+     */
+    class Position {
+    public:
+        // This represents "not found" similar to string::npos
+        Position() :index(static_cast<unsigned>(-1)) {}
+        bool found() const { return index != Position().index; }
+
+        bool operator == (Position rhs) const { return this->index == rhs.index; }
+        bool operator != (Position rhs) const { return !(*this == rhs); }
+
+        // For debugging and ASSERT_EQUALS in tests.
+        template <typename OStream>
+        friend OStream& operator<<(OStream& stream, Position p) { return stream << p.index; }
+
+    private:
+        explicit Position(size_t i) :index(i) {}
+        unsigned index;
+        friend class DocumentStorage;
+        friend class DocumentStorageIterator;
+    };
+
+#pragma pack(1)
+    /** This is how values are stored in the DocumentStorage buffer
+     *  Internal class. Consumers shouldn't care about this.
+     */
+    class ValueElement : boost::noncopyable {
+    public:
+        Value val;
+        Position nextCollision; // Position of next field with same hashBucket
+        const int nameLen; // doesn't include '\0'
+        const char name[1]; // pointer to start of name
+
+        // implicit conversions to Value
+        operator Value&() { return val; }
+        operator const Value&() const  { return val; }
+
+        ValueElement* next() {
+            return align(plusBytes(sizeof(ValueElement) + nameLen));
+        }
+
+        const ValueElement* next() const {
+            return align(plusBytes(sizeof(ValueElement) + nameLen));
+        }
+
+        StringData nameSD() const { return StringData(name, nameLen); }
+
+
+        // helpers for doing pointer arithmetic with this class
+        // Note: These don't dereference 'this' so they are safe to use with NULL
+        char* ptr() { return reinterpret_cast<char*>(this); }
+        const char* ptr() const { return reinterpret_cast<const char*>(this); }
+        const ValueElement* plusBytes(size_t bytes) const {
+            return reinterpret_cast<const ValueElement*>(ptr() + bytes);
+        }
+        ValueElement* plusBytes(size_t bytes) {
+            return reinterpret_cast<ValueElement*>(ptr() + bytes);
+        }
+
+        // Round number or pointer up to N-byte boundary. No change if already aligned.
+        template <typename T>
+        static T align(T size) {
+            const intmax_t ALIGNMENT = 8; // must be power of 2 and <= 16 (malloc alignment)
+            // Can't use c++ cast because of conversion between intmax_t and both ints and pointers
+            return (T)(((intmax_t)(size) + (ALIGNMENT-1)) & ~(ALIGNMENT-1));
+        }
+
+    private:
+        ValueElement(); // this class should never be constructed
+        ~ValueElement(); // or destructed
+    };
+    // Real size is sizeof(ValueElement) + nameLen
+#pragma pack()
+    BOOST_STATIC_ASSERT(sizeof(ValueElement) == (sizeof(Value) +
+                                                 sizeof(Position) +
+                                                 sizeof(int) +
+                                                 1));
+
+    // This is an internal class for Document. See FieldIterator for the public version.
+    class DocumentStorageIterator {
+    public:
+        // DocumentStorage::iterator() and iteratorAll() are easier to use
+        DocumentStorageIterator(const ValueElement* first,
+                                const ValueElement* end,
+                                bool includeMissing)
+                : _first(first)
+                , _it(first)
+                , _end(end)
+                , _includeMissing(includeMissing) {
+            skipMissing();
+        }
+
+        bool atEnd() const { return _it == _end; }
+
+        const ValueElement& get() const { return *_it; }
+
+        Position position() const { return Position(_it->ptr() - _first->ptr()); }
+
+        void advance() {
+            advanceOne();
+            skipMissing();
+        }
+
+        const ValueElement* operator-> () { return _it; }
+        const ValueElement& operator* () { return *_it; }
+
+    private:
+        void advanceOne() {
+            _it = _it->next();
+        }
+
+        void skipMissing() {
+            if (!_includeMissing) {
+                while (!atEnd() && _it->val.missing()) {
+                    advanceOne();
+                }
+            }
+        }
+
+        const ValueElement* _first;
+        const ValueElement* _it;
+        const ValueElement* _end;
+        bool _includeMissing;
+    };
+
+    /// Storage class used by both Document and MutableDocument
+    class DocumentStorage :  public RefCountable {
+    public:
+        // Note: default constructor should zero-init to support emptyDoc()
+        DocumentStorage() : _buffer(NULL)
+                          , _bufferEnd(NULL)
+                          , _usedBytes(0)
+                          , _numFields(0)
+                          , _hashTabMask(0)
+        {}
+        ~DocumentStorage();
+
+        static const DocumentStorage& emptyDoc() {
+            static const char emptyBytes[sizeof(DocumentStorage)] = {0};
+            return *reinterpret_cast<const DocumentStorage*>(emptyBytes);
+        }
+
+        size_t size() const {
+            // can't use _numFields because it includes removed Fields
+            size_t count = 0;
+            for (DocumentStorageIterator it = iterator(); !it.atEnd(); it.advance())
+                count++;
+            return count;
+        }
+
+        /// Returns the position of the next field to be inserted
+        Position getNextPosition() const { return Position(_usedBytes); }
+
+        /// Returns the position of the named field (may be missing) or Position()
+        Position findField(StringData name) const;
+
+        // Document uses these
+        const ValueElement& getField(Position pos) const {
+            verify(pos.found());
+            return *(_firstElement->plusBytes(pos.index));
+        }
+        Value getField(StringData name) const {
+            Position pos = findField(name);
+            if (!pos.found())
+                return Value();
+            return getField(pos);
+        }
+
+        // MutableDocument uses these
+        ValueElement& getField(Position pos) {
+            verify(pos.found());
+            return *(_firstElement->plusBytes(pos.index));
+        }
+        Value& getField(StringData name) {
+            Position pos = findField(name);
+            if (!pos.found())
+                return appendField(name); // TODO: find a way to avoid hashing name twice
+            return getField(pos);
+        }
+
+        /// Adds a new field with missing Value at the end of the document
+        Value& appendField(StringData name);
+
+        /** Preallocates space for fields. Use this to attempt to prevent buffer growth.
+         *  This is only valid to call before anything is added to the document.
+         */
+        void reserveFields(size_t expectedFields);
+
+        /// This skips missing values
+        DocumentStorageIterator iterator() const {
+            return DocumentStorageIterator(_firstElement, end(), false);
+        }
+
+        /// This includes missing values
+        DocumentStorageIterator iteratorAll() const {
+            return DocumentStorageIterator(_firstElement, end(), true);
+        }
+
+        /// Shallow copy of this. Caller owns memory.
+        intrusive_ptr<DocumentStorage> clone() const;
+
+        size_t allocatedBytes() const {
+            return !_buffer ? 0 : (_bufferEnd - _buffer + hashTabBytes());
+        }
+
+    private:
+
+        /// Same as lastElement->next() or firstElement() if empty.
+        const ValueElement* end() const { return _firstElement->plusBytes(_usedBytes); }
+
+        /// Allocates space in _buffer. Copies existing data if there is any.
+        void alloc(unsigned newSize);
+
+        /// Call after adding field to _buffer and increasing _numFields
+        void addFieldToHashTable(Position pos);
+
+        // assumes _hashTabMask is (power of two) - 1
+        unsigned hashTabBuckets() const { return _hashTabMask + 1; }
+        unsigned hashTabBytes() const { return hashTabBuckets() * sizeof(Position); }
+
+        /// rehash on buffer growth if load-factor > .5 (attempt to keep lf < 1 when full)
+        bool needRehash() const { return _numFields*2 > hashTabBuckets(); }
+
+        /// Initialize empty hash table
+        void hashTabInit() { memset(_hashTab, -1, hashTabBytes()); }
+
+        static unsigned hashKey(StringData name) {
+            // TODO consider FNV-1a once we have a better benchmark corpus
+            unsigned out;
+            MurmurHash3_x86_32(name.data(), name.size(), 0, &out);
+            return out;
+        }
+
+        unsigned bucketForKey(StringData name) const {
+            return hashKey(name) & _hashTabMask;
+        }
+
+        /// Adds all fields to the hash table
+        void rehash() {
+            hashTabInit();
+            for (DocumentStorageIterator it = iteratorAll(); !it.atEnd(); it.advance())
+                addFieldToHashTable(it.position());
+        }
+
+        enum {
+            HASH_TAB_INIT_SIZE = 8, // must be power of 2
+            HASH_TAB_MIN = 4, // don't hash fields for docs smaller than this
+                              // set to 1 to always hash
+        };
+
+        // _buffer layout:
+        // -------------------------------------------------------------------------------
+        // | ValueElement1 Name1 | ValueElement2 Name2 | ... FREE SPACE ... | Hash Table |
+        // -------------------------------------------------------------------------------
+        //  ^ _buffer and _firstElement point here                           ^
+        //                                _bufferEnd and _hashTab point here ^
+        //
+        //
+        // When the buffer grows, the hash table moves to the new end.
+        union {
+            char* _buffer;
+            ValueElement* _firstElement;
+        };
+
+        union {
+            // pointer to "end" of _buffer element space and start of hash table (same position)
+            char* _bufferEnd;
+            Position* _hashTab; // table lazily initialized once _numFields == HASH_TAB_MIN
+        };
+
+        unsigned _usedBytes; // position where next field would start
+        unsigned _numFields; // this includes removed fields
+        unsigned _hashTabMask; // equal to hashTabBuckets()-1 but used more often
+        // When adding a field, make sure to update clone() method
+    };
+}
diff --git a/src/mongo/db/pipeline/document_source.h b/src/mongo/db/pipeline/document_source.h
index 7fcce36e91f..02595148076 100755
--- a/src/mongo/db/pipeline/document_source.h
+++ b/src/mongo/db/pipeline/document_source.h
@@ -88,9 +88,14 @@ namespace mongo {
 
         /** @returns the current Document without advancing.
          *
-         *  some implementations do the equivalent of verify(!eof()) so check eof() first
+         *  It is illegal to call this without first checking eof() == false or advance() == true.
+         *
+         *  While it is legal to call getCurrent() multiple times between calls to advance, and
+         *  you will get the same Document returned, some DocumentSources do expensive work in
+         *  getCurrent(). You are advised to cache the result if you plan to access it more than
+         *  once.
          */
-        virtual intrusive_ptr<Document> getCurrent() = 0;
+        virtual Document getCurrent() = 0;
 
         /**
          * Inform the source that it is no longer needed and may release its resources.  After
@@ -258,7 +263,7 @@ namespace mongo {
         virtual ~DocumentSourceBsonArray();
         virtual bool eof();
         virtual bool advance();
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
         virtual void setSource(DocumentSource *pSource);
 
         /**
@@ -301,7 +306,7 @@ namespace mongo {
         virtual ~DocumentSourceCommandShards();
         virtual bool eof();
         virtual bool advance();
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
         virtual void setSource(DocumentSource *pSource);
 
         /* convenient shorthand for a commonly used type */
@@ -335,9 +340,11 @@ namespace mongo {
          */
         void getNextDocument();
 
+        bool unstarted;
+        bool hasCurrent;
         bool newSource; // set to true for the first item of a new source
         intrusive_ptr<DocumentSourceBsonArray> pBsonSource;
-        intrusive_ptr<Document> pCurrent;
+        Document pCurrent;
         ShardOutput::const_iterator iterator;
         ShardOutput::const_iterator listEnd;
     };
@@ -369,7 +376,7 @@ namespace mongo {
         virtual ~DocumentSourceCursor();
         virtual bool eof();
         virtual bool advance();
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
         virtual void setSource(DocumentSource *pSource);
 
         /**
@@ -437,7 +444,9 @@ namespace mongo {
 
         void findNext();
 
-        intrusive_ptr<Document> pCurrent;
+        bool unstarted;
+        bool hasCurrent;
+        Document pCurrent;
 
         string ns; // namespace
 
@@ -482,7 +491,7 @@ namespace mongo {
         virtual ~DocumentSourceFilterBase();
         virtual bool eof();
         virtual bool advance();
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
 
         /**
           Create a BSONObj suitable for Matcher construction.
@@ -508,15 +517,15 @@ namespace mongo {
           @param pDocument the document to test
           @returns true if the document matches the filter, false otherwise
          */
-        virtual bool accept(const intrusive_ptr<Document> &pDocument) const = 0;
+        virtual bool accept(const Document& pDocument) const = 0;
 
     private:
 
         void findNext();
 
         bool unstarted;
-        bool hasNext;
-        intrusive_ptr<Document> pCurrent;
+        bool hasCurrent;
+        Document pCurrent;
     };
 
 
@@ -571,7 +580,7 @@ namespace mongo {
         virtual void sourceToBson(BSONObjBuilder *pBuilder, bool explain) const;
 
         // virtuals from DocumentSourceFilterBase
-        virtual bool accept(const intrusive_ptr<Document> &pDocument) const;
+        virtual bool accept(const Document& pDocument) const;
 
     private:
         DocumentSourceFilter(const intrusive_ptr<Expression> &pFilter,
@@ -589,7 +598,7 @@ namespace mongo {
         virtual bool eof();
         virtual bool advance();
         virtual const char *getSourceName() const;
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
         virtual GetDepsReturn getDependencies(set<string>& deps) const;
 
         /**
@@ -668,7 +677,7 @@ namespace mongo {
 
         intrusive_ptr<Expression> pIdExpression;
 
-        typedef boost::unordered_map<intrusive_ptr<const Value>,
+        typedef boost::unordered_map<Value,
             vector<intrusive_ptr<Accumulator> >, Value::Hash> GroupsType;
         GroupsType groups;
 
@@ -690,11 +699,10 @@ namespace mongo {
         vector<intrusive_ptr<Expression> > vpExpression;
 
 
-        intrusive_ptr<Document> makeDocument(
-            const GroupsType::iterator &rIter);
+        Document makeDocument(const GroupsType::iterator &rIter);
 
         GroupsType::iterator groupsIterator;
-        intrusive_ptr<Document> pCurrent;
+        Document pCurrent;
     };
 
 
@@ -735,7 +743,7 @@ namespace mongo {
         virtual void sourceToBson(BSONObjBuilder *pBuilder, bool explain) const;
 
         // virtuals from DocumentSourceFilterBase
-        virtual bool accept(const intrusive_ptr<Document> &pDocument) const;
+        virtual bool accept(const Document& pDocument) const;
 
     private:
         DocumentSourceMatch(const BSONObj &query,
@@ -753,7 +761,7 @@ namespace mongo {
         virtual bool eof();
         virtual bool advance();
         virtual const char *getSourceName() const;
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
 
         /**
           Create a document source for output and pass-through.
@@ -789,7 +797,7 @@ namespace mongo {
         virtual bool eof();
         virtual bool advance();
         virtual const char *getSourceName() const;
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
         virtual void optimize();
 
         virtual GetDepsReturn getDependencies(set<string>& deps) const;
@@ -839,7 +847,7 @@ namespace mongo {
         virtual bool eof();
         virtual bool advance();
         virtual const char *getSourceName() const;
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
 
         virtual GetDepsReturn getDependencies(set<string>& deps) const;
 
@@ -932,8 +940,7 @@ namespace mongo {
           @returns a number less than, equal to, or greater than zero,
             indicating pL < pR, pL == pR, or pL > pR, respectively
          */
-        int compare(const intrusive_ptr<Document> &pL,
-                    const intrusive_ptr<Document> &pR);
+        int compare(const Document& pL, const Document& pR);
 
         /*
           This is a utility class just for the STL sort that is done
@@ -941,9 +948,7 @@ namespace mongo {
          */
         class Comparator {
         public:
-            bool operator()(
-                const intrusive_ptr<Document> &pL,
-                const intrusive_ptr<Document> &pR) {
+            bool operator()(const Document& pL, const Document& pR) {
                 return (pSort->compare(pL, pR) < 0);
             }
 
@@ -955,11 +960,11 @@ namespace mongo {
             DocumentSourceSort *pSort;
         };
 
-        typedef vector<intrusive_ptr<Document> > VectorType;
+        typedef vector<Document> VectorType;
         VectorType documents;
 
         VectorType::iterator docIterator;
-        intrusive_ptr<Document> pCurrent;
+        Document pCurrent;
     };
 
 
@@ -970,7 +975,7 @@ namespace mongo {
         virtual ~DocumentSourceLimit();
         virtual bool eof();
         virtual bool advance();
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
         virtual const char *getSourceName() const;
         virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);
 
@@ -1022,7 +1027,6 @@ namespace mongo {
 
         long long limit;
         long long count;
-        intrusive_ptr<Document> pCurrent;
     };
 
     class DocumentSourceSkip :
@@ -1032,7 +1036,7 @@ namespace mongo {
         virtual ~DocumentSourceSkip();
         virtual bool eof();
         virtual bool advance();
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
         virtual const char *getSourceName() const;
         virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);
 
@@ -1088,7 +1092,7 @@ namespace mongo {
 
         long long skip;
         long long count;
-        intrusive_ptr<Document> pCurrent;
+        Document pCurrent;
     };
 
 
@@ -1100,7 +1104,7 @@ namespace mongo {
         virtual bool eof();
         virtual bool advance();
         virtual const char *getSourceName() const;
-        virtual intrusive_ptr<Document> getCurrent();
+        virtual Document getCurrent();
 
         virtual GetDepsReturn getDependencies(set<string>& deps) const;
 
diff --git a/src/mongo/db/pipeline/document_source_bson_array.cpp b/src/mongo/db/pipeline/document_source_bson_array.cpp
index 79bce29220a..64217cc9716 100755
--- a/src/mongo/db/pipeline/document_source_bson_array.cpp
+++ b/src/mongo/db/pipeline/document_source_bson_array.cpp
@@ -44,10 +44,10 @@ namespace mongo {
         return true;
     }
 
-    intrusive_ptr<Document> DocumentSourceBsonArray::getCurrent() {
+    Document DocumentSourceBsonArray::getCurrent() {
         verify(haveCurrent);
         BSONObj documentObj(currentElement.Obj());
-        intrusive_ptr<Document> pDocument(
+        Document pDocument(
             Document::createFromBsonObj(&documentObj));
         return pDocument;
     }
diff --git a/src/mongo/db/pipeline/document_source_command_shards.cpp b/src/mongo/db/pipeline/document_source_command_shards.cpp
index a0d5423b449..e1f04ae1a50 100644
--- a/src/mongo/db/pipeline/document_source_command_shards.cpp
+++ b/src/mongo/db/pipeline/document_source_command_shards.cpp
@@ -26,25 +26,25 @@ namespace mongo {
 
     bool DocumentSourceCommandShards::eof() {
         /* if we haven't even started yet, do so */
-        if (!pCurrent.get())
+        if (unstarted)
             getNextDocument();
 
-        return (pCurrent.get() == NULL);
+        return !hasCurrent;
     }
 
     bool DocumentSourceCommandShards::advance() {
         DocumentSource::advance(); // check for interrupts
 
-        if (eof())
-            return false;
+        if (unstarted)
+            getNextDocument(); // skip first
 
         /* advance */
         getNextDocument();
 
-        return (pCurrent.get() != NULL);
+        return hasCurrent;
     }
 
-    intrusive_ptr<Document> DocumentSourceCommandShards::getCurrent() {
+    Document DocumentSourceCommandShards::getCurrent() {
         verify(!eof());
         return pCurrent;
     }
@@ -64,6 +64,8 @@ namespace mongo {
         const ShardOutput& shardOutput,
         const intrusive_ptr<ExpressionContext> &pExpCtx):
         DocumentSource(pExpCtx),
+        unstarted(true),
+        hasCurrent(false),
         newSource(false),
         pBsonSource(),
         pCurrent(),
@@ -81,11 +83,17 @@ namespace mongo {
     }
 
     void DocumentSourceCommandShards::getNextDocument() {
+        if (unstarted) {
+            unstarted = false;
+            hasCurrent = true;
+        }
+
         while(true) {
             if (!pBsonSource.get()) {
                 /* if there aren't any more futures, we're done */
                 if (iterator == listEnd) {
-                    pCurrent.reset();
+                    pCurrent = Document();
+                    hasCurrent = false;
                     return;
                 }
 
diff --git a/src/mongo/db/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index c99504ad446..4e932069dbf 100755
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -37,28 +37,25 @@ namespace mongo {
 
     bool DocumentSourceCursor::eof() {
         /* if we haven't gotten the first one yet, do so now */
-        if (!pCurrent.get())
+        if (unstarted)
             findNext();
 
-        return (pCurrent.get() == NULL);
+        return !hasCurrent;
     }
 
     bool DocumentSourceCursor::advance() {
         DocumentSource::advance(); // check for interrupts
 
         /* if we haven't gotten the first one yet, do so now */
-        if (!pCurrent.get())
+        if (unstarted)
             findNext();
 
         findNext();
-        return (pCurrent.get() != NULL);
+        return hasCurrent;
     }
 
-    intrusive_ptr<Document> DocumentSourceCursor::getCurrent() {
-        /* if we haven't gotten the first one yet, do so now */
-        if (!pCurrent.get())
-            findNext();
-
+    Document DocumentSourceCursor::getCurrent() {
+        verify(hasCurrent);
         return pCurrent;
     }
 
@@ -98,9 +95,11 @@ namespace mongo {
     }
 
     void DocumentSourceCursor::findNext() {
+        unstarted = false;
 
         if ( !_cursorWithContext ) {
-            pCurrent.reset();
+            pCurrent = Document();
+            hasCurrent = false;
             return;
         }
 
@@ -135,6 +134,7 @@ namespace mongo {
             }
 
             pCurrent = Document::createFromBsonObj(&documentObj);
+            hasCurrent = true;
 
             cursor()->advance();
             return;
@@ -143,7 +143,8 @@ namespace mongo {
         // If we got here, there aren't any more documents.
         // The CursorWithContext (and its read lock) must be released, see SERVER-6123.
         dispose();
-        pCurrent.reset();
+        pCurrent = Document();
+        hasCurrent = false;
     }
 
     void DocumentSourceCursor::setSource(DocumentSource *pSource) {
@@ -193,7 +194,8 @@ namespace mongo {
         const shared_ptr<CursorWithContext>& cursorWithContext,
         const intrusive_ptr<ExpressionContext> &pCtx):
         DocumentSource(pCtx),
-        pCurrent(),
+        unstarted(true),
+        hasCurrent(false),
         _cursorWithContext( cursorWithContext )
     {}
 
diff --git a/src/mongo/db/pipeline/document_source_filter.cpp b/src/mongo/db/pipeline/document_source_filter.cpp
index b9cb0a369d1..de0736beee5 100755
--- a/src/mongo/db/pipeline/document_source_filter.cpp
+++ b/src/mongo/db/pipeline/document_source_filter.cpp
@@ -63,10 +63,9 @@ namespace mongo {
         pFilter->addToBsonObj(pBuilder, filterName, false);
     }
 
-    bool DocumentSourceFilter::accept(
-        const intrusive_ptr<Document> &pDocument) const {
-        intrusive_ptr<const Value> pValue(pFilter->evaluate(pDocument));
-        return pValue->coerceToBool();
+    bool DocumentSourceFilter::accept(const Document& pDocument) const {
+        Value pValue(pFilter->evaluate(pDocument));
+        return pValue.coerceToBool();
     }
 
     intrusive_ptr<DocumentSource> DocumentSourceFilter::createFromBson(
diff --git a/src/mongo/db/pipeline/document_source_filter_base.cpp b/src/mongo/db/pipeline/document_source_filter_base.cpp
index 3354b3c6bc2..fd42993b23e 100755
--- a/src/mongo/db/pipeline/document_source_filter_base.cpp
+++ b/src/mongo/db/pipeline/document_source_filter_base.cpp
@@ -28,30 +28,27 @@ namespace mongo {
     }
 
     void DocumentSourceFilterBase::findNext() {
-        /* only do this the first time */
-        if (unstarted) {
-            hasNext = !pSource->eof();
-            unstarted = false;
-        }
-
-        while(hasNext) {
-            boost::intrusive_ptr<Document> pDocument(pSource->getCurrent());
-            hasNext = pSource->advance();
+        unstarted = false;
 
-            if (accept(pDocument)) {
-                pCurrent = pDocument;
+        for(bool hasDoc = !pSource->eof(); hasDoc; hasDoc = pSource->advance()) {
+            pCurrent = pSource->getCurrent();
+            if (accept(pCurrent)) {
+                pSource->advance(); // Start next call at correct position
+                hasCurrent = true;
                 return;
             }
         }
 
-        pCurrent.reset();
+        // Nothing matched
+        pCurrent = Document();
+        hasCurrent = false;
     }
 
     bool DocumentSourceFilterBase::eof() {
         if (unstarted)
             findNext();
 
-        return (pCurrent.get() == NULL);
+        return !hasCurrent;
     }
 
     bool DocumentSourceFilterBase::advance() {
@@ -68,14 +65,11 @@ namespace mongo {
          */
         findNext();
 
-        return (pCurrent.get() != NULL);
+        return hasCurrent;
     }
 
-    boost::intrusive_ptr<Document> DocumentSourceFilterBase::getCurrent() {
-        if (unstarted)
-            findNext();
-
-        verify(pCurrent.get() != NULL);
+    Document DocumentSourceFilterBase::getCurrent() {
+        verify(hasCurrent);
         return pCurrent;
     }
 
@@ -83,7 +77,7 @@ namespace mongo {
         const intrusive_ptr<ExpressionContext> &pExpCtx):
         DocumentSource(pExpCtx),
         unstarted(true),
-        hasNext(false),
+        hasCurrent(false),
         pCurrent() {
     }
 }
diff --git a/src/mongo/db/pipeline/document_source_group.cpp b/src/mongo/db/pipeline/document_source_group.cpp
index c84d73d16b6..1fca0cd9467 100755
--- a/src/mongo/db/pipeline/document_source_group.cpp
+++ b/src/mongo/db/pipeline/document_source_group.cpp
@@ -60,7 +60,7 @@ namespace mongo {
         return true;
     }
 
-    intrusive_ptr<Document> DocumentSourceGroup::getCurrent() {
+    Document DocumentSourceGroup::getCurrent() {
         if (!populated)
             populate();
 
@@ -72,7 +72,7 @@ namespace mongo {
         BSONObjBuilder insides;
 
         /* add the _id */
-        pIdExpression->addToBsonObj(&insides, Document::idName.c_str(), true);
+        pIdExpression->addToBsonObj(&insides, "_id", true);
 
         /* add the remaining fields */
         const size_t n = vFieldName.size();
@@ -173,7 +173,7 @@ namespace mongo {
             BSONElement groupField(groupIterator.next());
             const char *pFieldName = groupField.fieldName();
 
-            if (strcmp(pFieldName, Document::idName.c_str()) == 0) {
+            if (str::equals(pFieldName, "_id")) {
                 uassert(15948, "a group's _id may only be specified once",
                         !idSet);
 
@@ -221,7 +221,7 @@ namespace mongo {
                     case jstNULL:
                     StringConstantId: // from string case above
                     {
-                        intrusive_ptr<const Value> pValue(
+                        Value pValue(
                             Value::createFromBsonElement(&groupField));
                         intrusive_ptr<ExpressionConstant> pConstant(
                             ExpressionConstant::create(pValue));
@@ -313,14 +313,14 @@ namespace mongo {
     void DocumentSourceGroup::populate() {
         for(bool hasNext = !pSource->eof(); hasNext;
                 hasNext = pSource->advance()) {
-            intrusive_ptr<Document> pDocument(pSource->getCurrent());
+            Document pDocument(pSource->getCurrent());
 
             /* get the _id value */
-            intrusive_ptr<const Value> pId(pIdExpression->evaluate(pDocument));
+            Value pId(pIdExpression->evaluate(pDocument));
 
             /* treat Undefined the same as NULL SERVER-4674 */
-            if (pId->getType() == Undefined)
-                pId = Value::getNull();
+            if (pId.getType() == Undefined)
+                pId = Value(jstNULL);
 
             /*
               Look for the _id value in the map; if it's not there, add a
@@ -334,10 +334,7 @@ namespace mongo {
             }
             else {
                 /* insert a new group into the map */
-                groups.insert(it,
-                              pair<intrusive_ptr<const Value>,
-                              vector<intrusive_ptr<Accumulator> > >(
-                                  pId, vector<intrusive_ptr<Accumulator> >()));
+                groups[pId] = vector<intrusive_ptr<Accumulator> >();
 
                 /* find the accumulator vector (the map value) */
                 it = groups.find(pId);
@@ -370,23 +367,23 @@ namespace mongo {
         populated = true;
     }
 
-    intrusive_ptr<Document> DocumentSourceGroup::makeDocument(
+    Document DocumentSourceGroup::makeDocument(
         const GroupsType::iterator &rIter) {
         vector<intrusive_ptr<Accumulator> > *pGroup = &rIter->second;
         const size_t n = vFieldName.size();
-        intrusive_ptr<Document> pResult(Document::create(1 + n));
+        MutableDocument out (1 + n);
 
         /* add the _id field */
-        pResult->addField(Document::idName, rIter->first);
+        out.addField("_id", rIter->first);
 
         /* add the rest of the fields */
         for(size_t i = 0; i < n; ++i) {
-            intrusive_ptr<const Value> pValue((*pGroup)[i]->getValue());
-            if (pValue->getType() != Undefined)
-                pResult->addField(vFieldName[i], pValue);
+            Value pValue((*pGroup)[i]->getValue());
+            if (pValue.getType() != Undefined)
+                out.addField(vFieldName[i], pValue);
         }
 
-        return pResult;
+        return out.freeze();
     }
 
     intrusive_ptr<DocumentSource> DocumentSourceGroup::getShardSource() {
@@ -399,8 +396,7 @@ namespace mongo {
         intrusive_ptr<DocumentSourceGroup> pMerger(DocumentSourceGroup::create(pMergerExpCtx));
 
         /* the merger will use the same grouping key */
-        pMerger->setIdExpression(ExpressionFieldPath::create(
-                                     Document::idName.c_str()));
+        pMerger->setIdExpression(ExpressionFieldPath::create("_id"));
 
         const size_t n = vFieldName.size();
         for(size_t i = 0; i < n; ++i) {
diff --git a/src/mongo/db/pipeline/document_source_limit.cpp b/src/mongo/db/pipeline/document_source_limit.cpp
index 999d80f0368..86dd4f22a5c 100644
--- a/src/mongo/db/pipeline/document_source_limit.cpp
+++ b/src/mongo/db/pipeline/document_source_limit.cpp
@@ -64,20 +64,16 @@ namespace mongo {
 
         ++count;
         if (count >= limit) {
-
-            pCurrent.reset();
-
             // This is required for the DocumentSourceCursor to release its read lock, see
             // SERVER-6123.
             pSource->dispose();
 
             return false;
         }
-        pCurrent = pSource->getCurrent();
         return pSource->advance();
     }
 
-    intrusive_ptr<Document> DocumentSourceLimit::getCurrent() {
+    Document DocumentSourceLimit::getCurrent() {
         return pSource->getCurrent();
     }
 
diff --git a/src/mongo/db/pipeline/document_source_match.cpp b/src/mongo/db/pipeline/document_source_match.cpp
index 03006c1e043..b8900bf19c8 100644
--- a/src/mongo/db/pipeline/document_source_match.cpp
+++ b/src/mongo/db/pipeline/document_source_match.cpp
@@ -40,8 +40,7 @@ namespace mongo {
         pBuilder->append(matchName, *pQuery);
     }
 
-    bool DocumentSourceMatch::accept(
-        const intrusive_ptr<Document> &pDocument) const {
+    bool DocumentSourceMatch::accept(const Document& pDocument) const {
 
         /*
           The matcher only takes BSON documents, so we have to make one.
diff --git a/src/mongo/db/pipeline/document_source_out.cpp b/src/mongo/db/pipeline/document_source_out.cpp
index c19e066efe6..bbdaacfa053 100755
--- a/src/mongo/db/pipeline/document_source_out.cpp
+++ b/src/mongo/db/pipeline/document_source_out.cpp
@@ -40,7 +40,7 @@ namespace mongo {
         return pSource->advance();
     }
 
-    boost::intrusive_ptr<Document> DocumentSourceOut::getCurrent() {
+    Document DocumentSourceOut::getCurrent() {
         return pSource->getCurrent();
     }
 
diff --git a/src/mongo/db/pipeline/document_source_project.cpp b/src/mongo/db/pipeline/document_source_project.cpp
index bbd969fa997..17fcb98c42d 100644
--- a/src/mongo/db/pipeline/document_source_project.cpp
+++ b/src/mongo/db/pipeline/document_source_project.cpp
@@ -48,13 +48,12 @@ namespace mongo {
         return pSource->advance();
     }
 
-    intrusive_ptr<Document> DocumentSourceProject::getCurrent() {
-        intrusive_ptr<Document> pInDocument(pSource->getCurrent());
-        verify(pInDocument);
+    Document DocumentSourceProject::getCurrent() {
+        Document pInDocument(pSource->getCurrent());
 
         /* create the result document */
         const size_t sizeHint = pEO->getSizeHint();
-        intrusive_ptr<Document> pResultDocument(Document::create(sizeHint));
+        MutableDocument out (sizeHint);
 
         /*
           Use the ExpressionObject to create the base result.
@@ -62,7 +61,7 @@ namespace mongo {
           If we're excluding fields at the top level, leave out the _id if
           it is found, because we took care of it above.
         */
-        pEO->addToDocument(pResultDocument, pInDocument, /*root=*/pInDocument);
+        pEO->addToDocument(out, pInDocument, /*root=*/pInDocument);
 
 #if defined(_DEBUG)
         if (!_simpleProjection.getSpec().isEmpty()) {
@@ -73,7 +72,7 @@ namespace mongo {
             BSONObj input = inputBuilder.done();
 
             BSONObjBuilder outputBuilder;
-            pResultDocument->toBson(&outputBuilder);
+            out.peek().toBson(&outputBuilder);
             BSONObj output = outputBuilder.done();
 
             BSONObj projected = _simpleProjection.transform(input);
@@ -88,7 +87,7 @@ namespace mongo {
         }
 #endif
 
-        return pResultDocument;
+        return out.freeze();
     }
 
     void DocumentSourceProject::optimize() {
diff --git a/src/mongo/db/pipeline/document_source_skip.cpp b/src/mongo/db/pipeline/document_source_skip.cpp
index 5024b817a90..833de29061a 100644
--- a/src/mongo/db/pipeline/document_source_skip.cpp
+++ b/src/mongo/db/pipeline/document_source_skip.cpp
@@ -87,7 +87,7 @@ namespace mongo {
         return pSource->advance();
     }
 
-    intrusive_ptr<Document> DocumentSourceSkip::getCurrent() {
+    Document DocumentSourceSkip::getCurrent() {
         skipper();
         return pCurrent;
     }
diff --git a/src/mongo/db/pipeline/document_source_sort.cpp b/src/mongo/db/pipeline/document_source_sort.cpp
index 4b911894cdb..8739874ccde 100755
--- a/src/mongo/db/pipeline/document_source_sort.cpp
+++ b/src/mongo/db/pipeline/document_source_sort.cpp
@@ -61,7 +61,7 @@ namespace mongo {
         return true;
     }
 
-    intrusive_ptr<Document> DocumentSourceSort::getCurrent() {
+    Document DocumentSourceSort::getCurrent() {
         if (!populated)
             populate();
 
@@ -164,7 +164,7 @@ namespace mongo {
         /* pull everything from the underlying source */
         for(bool hasNext = !pSource->eof(); hasNext;
             hasNext = pSource->advance()) {
-            intrusive_ptr<Document> pDocument(pSource->getCurrent());
+            Document pDocument(pSource->getCurrent());
             documents.push_back(pDocument);
 
             dmm.addToTotal(pDocument->getApproximateSize());
@@ -182,8 +182,7 @@ namespace mongo {
         populated = true;
     }
 
-    int DocumentSourceSort::compare(
-        const intrusive_ptr<Document> &pL, const intrusive_ptr<Document> &pR) {
+    int DocumentSourceSort::compare(const Document& pL, const Document& pR) {
 
         /*
           populate() already checked that there is a non-empty sort key,
@@ -196,8 +195,8 @@ namespace mongo {
         for(size_t i = 0; i < n; ++i) {
             /* evaluate the sort keys */
             ExpressionFieldPath *pE = vSortKey[i].get();
-            intrusive_ptr<const Value> pLeft(pE->evaluate(pL));
-            intrusive_ptr<const Value> pRight(pE->evaluate(pR));
+            Value pLeft(pE->evaluate(pL));
+            Value pRight(pE->evaluate(pR));
 
             /*
               Compare the two values; if they differ, return.  If they are
diff --git a/src/mongo/db/pipeline/document_source_unwind.cpp b/src/mongo/db/pipeline/document_source_unwind.cpp
index 848251b350a..4086372133c 100755
--- a/src/mongo/db/pipeline/document_source_unwind.cpp
+++ b/src/mongo/db/pipeline/document_source_unwind.cpp
@@ -30,63 +30,53 @@ namespace mongo {
         /** @param unwindPath is the field path to the array to unwind. */
         Unwinder(const FieldPath& unwindPath);
         /** Reset the unwinder to unwind a new document. */
-        void resetDocument(const intrusive_ptr<Document>& document);
+        void resetDocument(const Document& document);
         /** @return true if done unwinding the last document passed to resetDocument(). */
         bool eof() const;
-        /**
-         * Try to advance to the next document unwound from the document passed to resetDocument().
-         * @return true if advanced to a new unwound document, but false if done advancing.
-         */
+        /** Try to advance to the next document unwound from the document passed to resetDocument().  */
         void advance();
         /**
          * @return the current document unwound from the document provided to resetDocument(), using
          * the current value in the array located at the provided unwindPath.  But @return
-         * intrusive_ptr<Document>() if resetDocument() has not been called or the results to unwind
+         * Document() if resetDocument() has not been called or the results to unwind
          * have been exhausted.
          */
-        intrusive_ptr<Document> getCurrent() const;
+        Document getCurrent();
     private:
-        /**
-         * @return the value at the unwind path, otherwise an empty pointer if no such value
-         * exists.  The _unwindPathFieldIndexes attribute will be set as the field path is traversed
-         * to find the value to unwind.
-         */
-        intrusive_ptr<const Value> extractUnwindValue();
         // Path to the array to unwind.
-        FieldPath _unwindPath;
-        // The souce document to unwind.
-        intrusive_ptr<Document> _document;
+        const FieldPath _unwindPath;
+
+        Value _inputArray;
+        MutableDocument _output;
+
         // Document indexes of the field path components.
-        vector<int> _unwindPathFieldIndexes;
-        // Iterator over the array within _document to unwind.
-        intrusive_ptr<ValueIterator> _unwindArrayIterator;
-        // The last value returned from _unwindArrayIterator.
-        intrusive_ptr<const Value> _unwindArrayIteratorCurrent;
+        vector<Position> _unwindPathFieldIndexes;
+        // Index into the _inputArray to return next.
+        size_t _index;
     };
 
     DocumentSourceUnwind::Unwinder::Unwinder(const FieldPath& unwindPath):
         _unwindPath(unwindPath) {
     }
 
-    void DocumentSourceUnwind::Unwinder::resetDocument(const intrusive_ptr<Document>& document) {
-        verify( document );
+    void DocumentSourceUnwind::Unwinder::resetDocument(const Document& document) {
 
         // Reset document specific attributes.
-        _document = document;
+        _inputArray = Value();
+        _output.reset(document);
         _unwindPathFieldIndexes.clear();
-        _unwindArrayIterator.reset();
-        _unwindArrayIteratorCurrent.reset();
+        _index = 0;
 
-        intrusive_ptr<const Value> pathValue = extractUnwindValue(); // sets _unwindPathFieldIndexes
-        if (!pathValue) {
+        Value pathValue = document.getNestedField(_unwindPath, &_unwindPathFieldIndexes);
+        if (pathValue.missing()) {
             // The path does not exist.
             return;
         }
 
         bool nothingToEmit =
-            (pathValue->getType() == jstNULL) ||
-            (pathValue->getType() == Undefined) ||
-            ((pathValue->getType() == Array) && (pathValue->getArrayLength() == 0));
+            (pathValue.getType() == jstNULL) ||
+            (pathValue.getType() == Undefined) ||
+            ((pathValue.getType() == Array) && (pathValue.getArrayLength() == 0));
 
         if (nothingToEmit) {
             // The target field exists, but there are no values to unwind.
@@ -96,100 +86,39 @@ namespace mongo {
         // The target field must be an array to unwind.
         uassert(15978, str::stream() << (string)DocumentSourceUnwind::unwindName
                 << ":  value at end of field path must be an array",
-                pathValue->getType() == Array);
+                pathValue.getType() == Array);
 
-        // Start the iterator used to unwind the array.
-        _unwindArrayIterator = pathValue->getArray();
-        verify(_unwindArrayIterator->more()); // Checked above that the array is nonempty.
-        // Pull the first value out of the iterator.
-        _unwindArrayIteratorCurrent = _unwindArrayIterator->next();
+        _inputArray = pathValue;
+        verify(!eof()); // Checked above that the array is nonempty.
     }
 
     bool DocumentSourceUnwind::Unwinder::eof() const {
-        return !_unwindArrayIteratorCurrent;
+        return (_inputArray.missing()) // getType asserts if missing
+            || (_inputArray.getType() != Array)
+            || (_index == _inputArray.getArrayLength());
     }
 
     void DocumentSourceUnwind::Unwinder::advance() {
-        if (!_unwindArrayIterator) {
-            // resetDocument() has not been called or the supplied document had no results to
-            // unwind.
-            _unwindArrayIteratorCurrent = NULL;
-        }
-        else if (!_unwindArrayIterator->more()) {
-            // There are no more results to unwind.
-            _unwindArrayIteratorCurrent = NULL;
-        }
-        else {
-            _unwindArrayIteratorCurrent = _unwindArrayIterator->next();
+        if (!eof()) { // don't advance past end()
+            _index++;
         }
     }
 
-    intrusive_ptr<Document> DocumentSourceUnwind::Unwinder::getCurrent() const {
-        if (!_unwindArrayIteratorCurrent) {
-            return NULL;
-        }
-
-        // Clone all the documents along the field path so that the end values are not shared across
-        // documents that have come out of this pipeline operator.  This is a partial deep clone.
-        // Because the value at the end will be replaced, everything along the path leading to that
-        // will be replaced in order not to share that change with any other clones (or the
-        // original).
-
-        intrusive_ptr<Document> clone(_document->clone());
-        intrusive_ptr<Document> current(clone);
-        const size_t n = _unwindPathFieldIndexes.size();
-        verify(n);
-        for(size_t i = 0; i < n; ++i) {
-            const size_t fi = _unwindPathFieldIndexes[i];
-            Document::FieldPair fp(current->getField(fi));
-            if (i + 1 < n) {
-                // For every object in the path but the last, clone it and continue on down.
-                intrusive_ptr<Document> next = fp.second->getDocument()->clone();
-                current->setField(fi, fp.first, Value::createDocument(next));
-                current = next;
-            }
-            else {
-                // In the last nested document, subsitute the current unwound value.
-                current->setField(fi, fp.first, _unwindArrayIteratorCurrent);
-            }
+    Document DocumentSourceUnwind::Unwinder::getCurrent() {
+        if (eof()) {
+            return Document();
         }
 
-        return clone;
-    }
+        // If needed, this will automatically clone all the documents along the
+        // field path so that the end values are not shared across documents
+        // that have come out of this pipeline operator.  This is a partial deep
+        // clone. Because the value at the end will be replaced, everything
+        // along the path leading to that will be replaced in order not to share
+        // that change with any other clones (or the original).
 
-    intrusive_ptr<const Value> DocumentSourceUnwind::Unwinder::extractUnwindValue() {
+        _output.setNestedField(_unwindPathFieldIndexes, _inputArray[_index]);
 
-        intrusive_ptr<Document> current = _document;
-        intrusive_ptr<const Value> pathValue;
-        const size_t pathLength = _unwindPath.getPathLength();
-        for(size_t i = 0; i < pathLength; ++i) {
-
-            size_t idx = current->getFieldIndex(_unwindPath.getFieldName(i));
-
-            if (idx == current->getFieldCount()) {
-                // The target field is missing.
-                return NULL;
-            }
-
-            // Record the indexes of the fields down the field path in order to quickly replace them
-            // as the documents along the field path are cloned.
-            _unwindPathFieldIndexes.push_back(idx);
-
-            pathValue = current->getField(idx).second;
-
-            if (i < pathLength - 1) {
-
-                if (pathValue->getType() != Object) {
-                    // The next field in the path cannot exist (inside a non object).
-                    return NULL;
-                }
-
-                // Move down the object tree.
-                current = pathValue->getDocument();
-            }
-        }
-
-        return pathValue;
+        return _output.peek();
     }
 
     const char DocumentSourceUnwind::unwindName[] = "$unwind";
@@ -248,8 +177,8 @@ namespace mongo {
         return !_unwinder->eof();
     }
 
-    intrusive_ptr<Document> DocumentSourceUnwind::getCurrent() {
-        lazyInit();
+    Document DocumentSourceUnwind::getCurrent() {
+        verify(!eof());
         return _unwinder->getCurrent();
     }
 
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index 912c251bc76..bf9586ad551 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -326,8 +326,7 @@ namespace mongo {
         return pExpression;
     }
 
-    intrusive_ptr<const Value> ExpressionAdd::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionAdd::evaluate(const Document& pDocument) const {
 
         /*
           We'll try to return the narrowest possible result value.  To do that
@@ -341,17 +340,17 @@ namespace mongo {
 
         const size_t n = vpOperand.size();
         for (size_t i = 0; i < n; ++i) {
-            intrusive_ptr<const Value> pValue(vpOperand[i]->evaluate(pDocument));
+            Value pValue(vpOperand[i]->evaluate(pDocument));
 
-            BSONType valueType = pValue->getType();
+            BSONType valueType = pValue.getType();
             uassert(16415, "$add does not support dates",
                     valueType != Date);
             uassert(16416, "$add does not support strings",
                     valueType != String);
 
-            totalType = Value::getWidestNumeric(totalType, pValue->getType());
-            doubleTotal += pValue->coerceToDouble();
-            longTotal += pValue->coerceToLong();
+            totalType = Value::getWidestNumeric(totalType, pValue.getType());
+            doubleTotal += pValue.coerceToDouble();
+            longTotal += pValue.coerceToLong();
         }
 
         if (totalType == NumberLong) {
@@ -417,10 +416,10 @@ namespace mongo {
           Evaluate and coerce the last argument to a boolean.  If it's false,
           then we can replace this entire expression.
          */
-        bool last = pLast->evaluate(intrusive_ptr<Document>())->coerceToBool();
+        bool last = pLast->evaluate(Document()).coerceToBool();
         if (!last) {
             intrusive_ptr<ExpressionConstant> pFinal(
-                ExpressionConstant::create(Value::getFalse()));
+                ExpressionConstant::create(Value(false)));
             return pFinal;
         }
 
@@ -447,16 +446,15 @@ namespace mongo {
         return pE;
     }
 
-    intrusive_ptr<const Value> ExpressionAnd::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionAnd::evaluate(const Document& pDocument) const {
         const size_t n = vpOperand.size();
         for(size_t i = 0; i < n; ++i) {
-            intrusive_ptr<const Value> pValue(vpOperand[i]->evaluate(pDocument));
-            if (!pValue->coerceToBool())
-                return Value::getFalse();
+            Value pValue(vpOperand[i]->evaluate(pDocument));
+            if (!pValue.coerceToBool())
+                return Value(false);
         }
 
-        return Value::getTrue();
+        return Value(true);
     }
 
     const char *ExpressionAnd::getOpName() const {
@@ -517,19 +515,17 @@ namespace mongo {
         pExpression->addDependencies(deps);
     }
 
-    intrusive_ptr<const Value> ExpressionCoerceToBool::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
-
-        intrusive_ptr<const Value> pResult(pExpression->evaluate(pDocument));
-        bool b = pResult->coerceToBool();
+    Value ExpressionCoerceToBool::evaluate(const Document& pDocument) const {
+        Value pResult(pExpression->evaluate(pDocument));
+        bool b = pResult.coerceToBool();
         if (b)
-            return Value::getTrue();
-        return Value::getFalse();
+            return Value(true);
+        return Value(false);
     }
 
-    void ExpressionCoerceToBool::addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const {
+    void ExpressionCoerceToBool::addToBsonObj(BSONObjBuilder *pBuilder,
+                                              StringData fieldName,
+                                              bool requireExpression) const {
         // Serializing as an $and expression which will become a CoerceToBool
         BSONObjBuilder sub (pBuilder->subobjStart(fieldName));
         BSONArrayBuilder arr (sub.subarrayStart("$and"));
@@ -681,33 +677,27 @@ namespace mongo {
             pFieldPath, newOp, pConstant->getValue());
     }
 
-    intrusive_ptr<const Value> ExpressionCompare::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionCompare::evaluate(const Document& pDocument) const {
         checkArgCount(2);
-        intrusive_ptr<const Value> pLeft(vpOperand[0]->evaluate(pDocument));
-        intrusive_ptr<const Value> pRight(vpOperand[1]->evaluate(pDocument));
+        Value pLeft(vpOperand[0]->evaluate(pDocument));
+        Value pRight(vpOperand[1]->evaluate(pDocument));
 
         int cmp = signum(Value::compare(pLeft, pRight));
 
         if (cmpOp == CMP) {
             switch(cmp) {
             case -1:
-                return Value::getMinusOne();
             case 0:
-                return Value::getZero();
             case 1:
-                return Value::getOne();
+                return Value(cmp);
 
             default:
                 verify(false); // CW TODO internal error
-                return Value::getNull();
             }
         }
 
         bool returnValue = cmpLookup[cmpOp].truthValue[cmp + 1];
-        if (returnValue)
-            return Value::getTrue();
-        return Value::getFalse();
+        return Value(returnValue);
     }
 
     const char *ExpressionCompare::getOpName() const {
@@ -734,11 +724,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionCond::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionCond::evaluate(const Document& pDocument) const {
         checkArgCount(3);
-        intrusive_ptr<const Value> pCond(vpOperand[0]->evaluate(pDocument));
-        int idx = pCond->coerceToBool() ? 1 : 2;
+        Value pCond(vpOperand[0]->evaluate(pDocument));
+        int idx = pCond.coerceToBool() ? 1 : 2;
         return vpOperand[idx]->evaluate(pDocument);
     }
 
@@ -762,16 +751,12 @@ namespace mongo {
         pValue(Value::createFromBsonElement(pBsonElement)) {
     }
 
-    intrusive_ptr<ExpressionConstant> ExpressionConstant::create(
-        const intrusive_ptr<const Value> &pValue) {
+    intrusive_ptr<ExpressionConstant> ExpressionConstant::create(const Value& pValue) {
         intrusive_ptr<ExpressionConstant> pEC(new ExpressionConstant(pValue));
         return pEC;
     }
 
-    ExpressionConstant::ExpressionConstant(
-        const intrusive_ptr<const Value> &pTheValue):
-        pValue(pTheValue) {
-    }
+    ExpressionConstant::ExpressionConstant(const Value& pTheValue): pValue(pTheValue) {}
 
 
     intrusive_ptr<Expression> ExpressionConstant::optimize() {
@@ -783,14 +768,13 @@ namespace mongo {
         /* nothing to do */
     }
 
-    intrusive_ptr<const Value> ExpressionConstant::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionConstant::evaluate(const Document& pDocument) const {
         return pValue;
     }
 
-    void ExpressionConstant::addToBsonObj(
-        BSONObjBuilder *pBuilder, const std::string& fieldName,
-        bool requireExpression) const {
+    void ExpressionConstant::addToBsonObj(BSONObjBuilder *pBuilder,
+                                          StringData fieldName,
+                                          bool requireExpression) const {
         /*
           If we don't need an expression, but can use a naked scalar,
           do the regular thing.
@@ -816,19 +800,19 @@ namespace mongo {
           which will look like an inclusion rather than a computed field.
         */
         if (!requireExpression) {
-            pValue->addToBsonObj(pBuilder, fieldName);
+            pValue.addToBsonObj(pBuilder, fieldName);
             return;
         }
 
         // We require an expression, so build one here, and use that.
         BSONObjBuilder constBuilder (pBuilder->subobjStart(fieldName));
-        pValue->addToBsonObj(&constBuilder, getOpName());
+        pValue.addToBsonObj(&constBuilder, getOpName());
         constBuilder.done();
     }
 
     void ExpressionConstant::addToBsonArray(
         BSONArrayBuilder *pBuilder) const {
-        pValue->addToBsonArray(pBuilder);
+        pValue.addToBsonArray(pBuilder);
     }
 
     const char *ExpressionConstant::getOpName() const {
@@ -855,11 +839,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionDayOfMonth::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionDayOfMonth::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_mday); 
     }
 
@@ -886,11 +869,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionDayOfWeek::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionDayOfWeek::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_wday+1); // MySQL uses 1-7 tm uses 0-6
     }
 
@@ -917,11 +899,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionDayOfYear::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionDayOfYear::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_yday+1); // MySQL uses 1-366 tm uses 0-365
     }
 
@@ -949,21 +930,20 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionDivide::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionDivide::evaluate(const Document& pDocument) const {
         checkArgCount(2);
-        intrusive_ptr<const Value> pLeft(vpOperand[0]->evaluate(pDocument));
-        intrusive_ptr<const Value> pRight(vpOperand[1]->evaluate(pDocument));
+        Value pLeft(vpOperand[0]->evaluate(pDocument));
+        Value pRight(vpOperand[1]->evaluate(pDocument));
 
         uassert(16373,
                 "$divide does not support dates",
-                pLeft->getType() != Date && pRight->getType() != Date);
+                pLeft.getType() != Date && pRight.getType() != Date);
 
-        double right = pRight->coerceToDouble();
+        double right = pRight.coerceToDouble();
         if (right == 0)
-            return Value::getUndefined();
+            return Value(Undefined);
 
-        double left = pLeft->coerceToDouble();
+        double left = pLeft.coerceToDouble();
 
         return Value::createDouble(left / right);
     }
@@ -1037,9 +1017,9 @@ namespace mongo {
     }
 
     void ExpressionObject::addToDocument(
-        const intrusive_ptr<Document> &pResult,
-        const intrusive_ptr<Document> &pDocument,
-        const intrusive_ptr<Document> &rootDoc
+        MutableDocument& out,
+        const Document& pDocument,
+        const Document& rootDoc
         ) const
     {
         const bool atRoot = (pDocument == rootDoc);
@@ -1053,14 +1033,16 @@ namespace mongo {
         while(fields.more()) {
             Document::FieldPair field (fields.next());
 
-            ExpressionMap::const_iterator exprIter = _expressions.find(field.first);
+            // TODO don't make a new string here
+            const string fieldName (field.first.data(), field.first.size());
+            ExpressionMap::const_iterator exprIter = _expressions.find(fieldName);
 
             // This field is not supposed to be in the output (unless it is _id)
             if (exprIter == end) {
-                if (!_excludeId && atRoot && field.first == Document::idName) {
+                if (!_excludeId && atRoot && str::equals(field.first.data(), "_id")) {
                     // _id from the root doc is always included (until exclusion is supported)
                     // not updating doneFields since "_id" isn't in _expressions
-                    pResult->addField(field.first, field.second);
+                    out.addField(field.first, field.second);
                 }
                 continue;
             }
@@ -1072,19 +1054,19 @@ namespace mongo {
 
             if (!expr) {
                 // This means pull the matching field from the input document
-                pResult->addField(field.first, field.second);
+                out.addField(field.first, field.second);
                 continue;
             }
 
             ExpressionObject* exprObj = dynamic_cast<ExpressionObject*>(expr);
-            BSONType valueType = field.second->getType();
+            BSONType valueType = field.second.getType();
             if ((valueType != Object && valueType != Array) || !exprObj ) {
                 // This expression replace the whole field
                 
-                intrusive_ptr<const Value> pValue(expr->evaluate(rootDoc));
+                Value pValue(expr->evaluate(rootDoc));
 
                 // don't add field if nothing was found in the subobject
-                if (exprObj && pValue->getDocument()->getFieldCount() == 0)
+                if (exprObj && pValue.getDocument()->getFieldCount() == 0)
                     continue;
 
                 /*
@@ -1093,8 +1075,8 @@ namespace mongo {
                    force the appearnance of non-existent fields.
                    */
                 // TODO make missing distinct from Undefined
-                if (pValue->getType() != Undefined)
-                    pResult->addField(field.first, pValue);
+                if (pValue.getType() != Undefined)
+                    out.addField(field.first, pValue);
 
 
                 continue;
@@ -1106,11 +1088,9 @@ namespace mongo {
                 add it to the result.
             */
             if (valueType == Object) {
-                intrusive_ptr<Document> doc = Document::create(exprObj->getSizeHint());
-                exprObj->addToDocument(doc,
-                                       field.second->getDocument(),
-                                       rootDoc);
-                pResult->addField(field.first, Value::createDocument(doc));
+                MutableDocument sub (exprObj->getSizeHint());
+                exprObj->addToDocument(sub, field.second.getDocument(), rootDoc);
+                out.addField(field.first, Value(sub.freeze()));
             }
             else if (valueType == Array) {
                 /*
@@ -1118,24 +1098,19 @@ namespace mongo {
                     but to each array element.  Then, add the array
                     of results to the current document.
                 */
-                vector<intrusive_ptr<const Value> > result;
-                intrusive_ptr<ValueIterator> pVI(field.second->getArray());
-                while(pVI->more()) {
-                    intrusive_ptr<const Value> next =  pVI->next();
-
+                vector<Value> result;
+                const vector<Value>& input = field.second.getArray();
+                for (size_t i=0; i < input.size(); i++) {
                     // can't look for a subfield in a non-object value.
-                    if (next->getType() != Object)
+                    if (input[i].getType() != Object)
                         continue;
 
-                    intrusive_ptr<Document> doc = Document::create(exprObj->getSizeHint());
-                    exprObj->addToDocument(doc,
-                                           next->getDocument(),
-                                           rootDoc);
-                    result.push_back(Value::createDocument(doc));
+                    MutableDocument doc (exprObj->getSizeHint());
+                    exprObj->addToDocument(doc, input[i].getDocument(), rootDoc);
+                    result.push_back(Value(doc.freeze()));
                 }
 
-                pResult->addField(field.first,
-                                    Value::createArray(result));
+                out.addField(field.first, Value(result));
             }
             else {
                 verify( false );
@@ -1158,23 +1133,23 @@ namespace mongo {
             if (!it->second)
                 continue;
 
-            intrusive_ptr<const Value> pValue(it->second->evaluate(rootDoc));
+            Value pValue(it->second->evaluate(rootDoc));
 
             /*
               Don't add non-existent values (note:  different from NULL);
               this is consistent with existing selection syntax which doesn't
               force the appearnance of non-existent fields.
             */
-            if (pValue->getType() == Undefined)
+            if (pValue.getType() == Undefined)
                 continue;
 
             // don't add field if nothing was found in the subobject
             if (dynamic_cast<ExpressionObject*>(it->second.get())
-                    && pValue->getDocument()->getFieldCount() == 0)
+                    && pValue.getDocument()->getFieldCount() == 0)
                 continue;
 
 
-            pResult->addField(fieldName, pValue);
+            out.addField(fieldName, pValue);
         }
     }
 
@@ -1183,20 +1158,17 @@ namespace mongo {
         return _expressions.size() + (_excludeId ? 0 : 1);
     }
 
-    intrusive_ptr<Document> ExpressionObject::evaluateDocument(
-        const intrusive_ptr<Document> &pDocument) const {
+    Document ExpressionObject::evaluateDocument(const Document& pDocument) const {
         /* create and populate the result */
-        intrusive_ptr<Document> pResult(
-            Document::create(getSizeHint()));
+        MutableDocument out (getSizeHint());
         
-        addToDocument(pResult,
-                      Document::create(), // No inclusion field matching.
+        addToDocument(out,
+                      Document(), // No inclusion field matching.
                       pDocument);
-        return pResult;
+        return out.freeze();
     }
 
-    intrusive_ptr<const Value> ExpressionObject::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionObject::evaluate(const Document& pDocument) const {
         return Value::createDocument(evaluateDocument(pDocument));
     }
 
@@ -1277,9 +1249,9 @@ namespace mongo {
         }
     }
 
-    void ExpressionObject::addToBsonObj(
-        BSONObjBuilder *pBuilder, const std::string& fieldName,
-        bool requireExpression) const {
+    void ExpressionObject::addToBsonObj(BSONObjBuilder *pBuilder,
+                                        StringData fieldName,
+                                        bool requireExpression) const {
 
         BSONObjBuilder objBuilder (pBuilder->subobjStart(fieldName));
         documentToBson(&objBuilder, requireExpression);
@@ -1325,16 +1297,16 @@ namespace mongo {
         deps.insert(fieldPath.getPath(false));
     }
 
-    intrusive_ptr<const Value> ExpressionFieldPath::evaluatePath(
-        size_t index, const size_t pathLength,
-        intrusive_ptr<Document> pDocument) const {
-        intrusive_ptr<const Value> pValue; /* the return value */
+    Value ExpressionFieldPath::evaluatePath(size_t index,
+                                            size_t pathLength,
+                                            Document pDocument) const {
+        Value pValue; /* the return value */
 
         pValue = pDocument->getValue(fieldPath.getFieldName(index));
 
         /* if the field doesn't exist, quit with an undefined value */
-        if (!pValue.get())
-            return Value::getUndefined();
+        if (pValue.missing())
+            return Value(Undefined);
 
         /* if we've hit the end of the path, stop */
         ++index;
@@ -1344,13 +1316,13 @@ namespace mongo {
         /*
           We're diving deeper.  If the value was null, return null.
         */
-        BSONType type = pValue->getType();
+        BSONType type = pValue.getType();
         if ((type == Undefined) || (type == jstNULL))
-            return Value::getUndefined();
+            return Value(Undefined);
 
         if (type == Object) {
             /* extract from the next level down */
-            return evaluatePath(index, pathLength, pValue->getDocument());
+            return evaluatePath(index, pathLength, pValue.getDocument());
         }
 
         if (type == Array) {
@@ -1358,13 +1330,13 @@ namespace mongo {
               We're going to repeat this for each member of the array,
               building up a new array as we go.
             */
-            vector<intrusive_ptr<const Value> > result;
-            intrusive_ptr<ValueIterator> pIter(pValue->getArray());
-            while(pIter->more()) {
-                intrusive_ptr<const Value> pItem(pIter->next());
-                BSONType iType = pItem->getType();
+            vector<Value> result;
+            const vector<Value>& input = pValue.getArray();
+            for (size_t i=0; i < input.size(); i++) {
+                const Value& item = input[i];
+                BSONType iType = item.getType();
                 if ((iType == Undefined) || (iType == jstNULL)) {
-                    result.push_back(pItem);
+                    result.push_back(item);
                     continue;
                 }
 
@@ -1374,25 +1346,23 @@ namespace mongo {
                         fieldPath.getPath(false) <<
                         "' is not an object, and cannot be navigated",
                         iType == Object);
-                intrusive_ptr<const Value> itemResult(
-                    evaluatePath(index, pathLength, pItem->getDocument()));
-                result.push_back(itemResult);
+
+                result.push_back(evaluatePath(index, pathLength, item.getDocument()));
             }
 
             return Value::createArray(result);
         }
         // subdocument field does not exist, return undefined
-        return Value::getUndefined();
+        return Value(Undefined);
     }
 
-    intrusive_ptr<const Value> ExpressionFieldPath::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionFieldPath::evaluate(const Document& pDocument) const {
         return evaluatePath(0, fieldPath.getPathLength(), pDocument);
     }
 
-    void ExpressionFieldPath::addToBsonObj(
-        BSONObjBuilder *pBuilder, const std::string& fieldName,
-        bool requireExpression) const {
+    void ExpressionFieldPath::addToBsonObj(BSONObjBuilder *pBuilder,
+                                           StringData fieldName,
+                                           bool requireExpression) const {
         pBuilder->append(fieldName, fieldPath.getPath(true));
     }
 
@@ -1409,15 +1379,15 @@ namespace mongo {
     intrusive_ptr<Expression> ExpressionFieldRange::optimize() {
         /* if there is no range to match, this will never evaluate true */
         if (!pRange.get())
-            return ExpressionConstant::create(Value::getFalse());
+            return ExpressionConstant::create(Value(false));
 
         /*
           If we ended up with a double un-ended range, anything matches.  I
           don't know how that can happen, given intersect()'s interface, but
           here it is, just in case.
         */
-        if (!pRange->pBottom.get() && !pRange->pTop.get())
-            return ExpressionConstant::create(Value::getTrue());
+        if (pRange->pBottom.missing() && pRange->pTop.missing())
+            return ExpressionConstant::create(Value(true));
 
         /*
           In all other cases, we have to test candidate values.  The
@@ -1431,20 +1401,19 @@ namespace mongo {
         pFieldPath->addDependencies(deps);
     }
 
-    intrusive_ptr<const Value> ExpressionFieldRange::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionFieldRange::evaluate(const Document& pDocument) const {
         /* if there's no range, there can't be a match */
         if (!pRange.get())
-            return Value::getFalse();
+            return Value(false);
 
         /* get the value of the specified field */
-        intrusive_ptr<const Value> pValue(pFieldPath->evaluate(pDocument));
+        Value pValue(pFieldPath->evaluate(pDocument));
 
         /* see if it fits within any of the ranges */
         if (pRange->contains(pValue))
-            return Value::getTrue();
+            return Value(true);
 
-        return Value::getFalse();
+        return Value(false);
     }
 
     void ExpressionFieldRange::addToBson(Builder *pBuilder) const {
@@ -1454,7 +1423,7 @@ namespace mongo {
             return;
         }
 
-        if (!pRange->pTop.get() && !pRange->pBottom.get()) {
+        if (pRange->pTop.missing() && pRange->pBottom.missing()) {
             /* any value will satisfy this predicate */
             pBuilder->append(true);
             return;
@@ -1463,10 +1432,10 @@ namespace mongo {
         // FIXME Append constant values using the $const operator.  SERVER-6769
 
         // FIXME This checks pointer equality not value equality.
-        if (pRange->pTop.get() == pRange->pBottom.get()) {
+        if (pRange->pTop == pRange->pBottom) {
             BSONArrayBuilder operands;
             pFieldPath->addToBsonArray(&operands);
-            pRange->pTop->addToBsonArray(&operands);
+            pRange->pTop.addToBsonArray(&operands);
             
             BSONObjBuilder equals;
             equals.append("$eq", operands.arr());
@@ -1475,30 +1444,30 @@ namespace mongo {
         }
 
         BSONObjBuilder leftOperator;
-        if (pRange->pBottom.get()) {
+        if (!pRange->pBottom.missing()) {
             BSONArrayBuilder leftOperands;
             pFieldPath->addToBsonArray(&leftOperands);
-            pRange->pBottom->addToBsonArray(&leftOperands);
+            pRange->pBottom.addToBsonArray(&leftOperands);
             leftOperator.append(
                 (pRange->bottomOpen ? "$gt" : "$gte"),
                 leftOperands.arr());
 
-            if (!pRange->pTop.get()) {
+            if (pRange->pTop.missing()) {
                 pBuilder->append(&leftOperator);
                 return;
             }
         }
 
         BSONObjBuilder rightOperator;
-        if (pRange->pTop.get()) {
+        if (!pRange->pTop.missing()) {
             BSONArrayBuilder rightOperands;
             pFieldPath->addToBsonArray(&rightOperands);
-            pRange->pTop->addToBsonArray(&rightOperands);
+            pRange->pTop.addToBsonArray(&rightOperands);
             rightOperator.append(
                 (pRange->topOpen ? "$lt" : "$lte"),
                 rightOperands.arr());
 
-            if (!pRange->pBottom.get()) {
+            if (pRange->pBottom.missing()) {
                 pBuilder->append(&rightOperator);
                 return;
             }
@@ -1512,9 +1481,9 @@ namespace mongo {
         pBuilder->append(&andOperator);
     }
 
-    void ExpressionFieldRange::addToBsonObj(
-        BSONObjBuilder *pBuilder, const std::string& fieldName,
-        bool requireExpression) const {
+    void ExpressionFieldRange::addToBsonObj(BSONObjBuilder *pBuilder,
+                                            StringData fieldName,
+                                            bool requireExpression) const {
         BuilderObj builder(pBuilder, fieldName);
         addToBson(&builder);
     }
@@ -1530,26 +1499,26 @@ namespace mongo {
         verify(pRange.get()); // otherwise, we can't do anything
 
         /* if there are no endpoints, then every value is accepted */
-        if (!pRange->pBottom.get() && !pRange->pTop.get())
+        if (pRange->pBottom.missing() && pRange->pTop.missing())
             return; // nothing to add to the predicate
 
         /* we're going to need the field path */
         string fieldPath(pFieldPath->getFieldPath(false));
 
         BSONObjBuilder range;
-        if (pRange->pBottom.get()) {
+        if (!pRange->pBottom.missing()) {
             /* the test for equality doesn't generate a subobject */
-            if (pRange->pBottom.get() == pRange->pTop.get()) {
-                pRange->pBottom->addToBsonObj(pBuilder, fieldPath);
+            if (pRange->pBottom == pRange->pTop) {
+                pRange->pBottom.addToBsonObj(pBuilder, fieldPath);
                 return;
             }
 
-            pRange->pBottom->addToBsonObj(
+            pRange->pBottom.addToBsonObj(
                 pBuilder, (pRange->bottomOpen ? "$gt" : "$gte"));
         }
 
-        if (pRange->pTop.get()) {
-            pRange->pTop->addToBsonObj(
+        if (!pRange->pTop.missing()) {
+            pRange->pTop.addToBsonObj(
                 pBuilder, (pRange->topOpen ? "$lt" : "$lte"));
         }
 
@@ -1558,7 +1527,7 @@ namespace mongo {
 
     intrusive_ptr<ExpressionFieldRange> ExpressionFieldRange::create(
         const intrusive_ptr<ExpressionFieldPath> &pFieldPath, CmpOp cmpOp,
-        const intrusive_ptr<const Value> &pValue) {
+        const Value& pValue) {
         intrusive_ptr<ExpressionFieldRange> pE(
             new ExpressionFieldRange(pFieldPath, cmpOp, pValue));
         return pE;
@@ -1566,13 +1535,12 @@ namespace mongo {
 
     ExpressionFieldRange::ExpressionFieldRange(
         const intrusive_ptr<ExpressionFieldPath> &pTheFieldPath, CmpOp cmpOp,
-        const intrusive_ptr<const Value> &pValue):
+        const Value& pValue):
         pFieldPath(pTheFieldPath),
         pRange(new Range(cmpOp, pValue)) {
     }
 
-    void ExpressionFieldRange::intersect(
-        CmpOp cmpOp, const intrusive_ptr<const Value> &pValue) {
+    void ExpressionFieldRange::intersect(CmpOp cmpOp, const Value& pValue) {
 
         /* create the new range */
         scoped_ptr<Range> pNew(new Range(cmpOp, pValue));
@@ -1586,8 +1554,7 @@ namespace mongo {
         pRange.reset(pRange->intersect(pNew.get()));
     }
 
-    ExpressionFieldRange::Range::Range(
-        CmpOp cmpOp, const intrusive_ptr<const Value> &pValue):
+    ExpressionFieldRange::Range::Range(CmpOp cmpOp, const Value& pValue):
         bottomOpen(false),
         topOpen(false),
         pBottom(),
@@ -1629,8 +1596,8 @@ namespace mongo {
     }
 
     ExpressionFieldRange::Range::Range(
-        const intrusive_ptr<const Value> &pTheBottom, bool theBottomOpen,
-        const intrusive_ptr<const Value> &pTheTop, bool theTopOpen):
+        const Value& pTheBottom, bool theBottomOpen,
+        const Value& pTheTop, bool theTopOpen):
         bottomOpen(theBottomOpen),
         topOpen(theTopOpen),
         pBottom(pTheBottom),
@@ -1645,10 +1612,10 @@ namespace mongo {
           Start by assuming the maximum is from pRange.  Then, if we have
           values of our own, see if they're greater.
         */
-        intrusive_ptr<const Value> pMaxBottom(pRange->pBottom);
+        Value pMaxBottom(pRange->pBottom);
         bool maxBottomOpen = pRange->bottomOpen;
-        if (pBottom.get()) {
-            if (!pRange->pBottom.get()) {
+        if (!pBottom.missing()) {
+            if (pRange->pBottom.missing()) {
                 pMaxBottom = pBottom;
                 maxBottomOpen = bottomOpen;
             }
@@ -1669,10 +1636,10 @@ namespace mongo {
           Start by assuming the minimum is from pRange.  Then, if we have
           values of our own, see if they are less.
         */
-        intrusive_ptr<const Value> pMinTop(pRange->pTop);
+        Value pMinTop(pRange->pTop);
         bool minTopOpen = pRange->topOpen;
-        if (pTop.get()) {
-            if (!pRange->pTop.get()) {
+        if (!pTop.missing()) {
+            if (pRange->pTop.missing()) {
                 pMinTop = pTop;
                 minTopOpen = topOpen;
             }
@@ -1698,9 +1665,8 @@ namespace mongo {
         return NULL;
     }
 
-    bool ExpressionFieldRange::Range::contains(
-        const intrusive_ptr<const Value> &pValue) const {
-        if (pBottom.get()) {
+    bool ExpressionFieldRange::Range::contains(const Value& pValue) const {
+        if (!pBottom.missing()) {
             const int cmp = Value::compare(pValue, pBottom);
             if (cmp < 0)
                 return false;
@@ -1708,7 +1674,7 @@ namespace mongo {
                 return false;
         }
 
-        if (pTop.get()) {
+        if (!pTop.missing()) {
             const int cmp = Value::compare(pValue, pTop);
             if (cmp > 0)
                 return false;
@@ -1739,11 +1705,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionMinute::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionMinute::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_min);
     }
 
@@ -1771,14 +1736,13 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionMod::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionMod::evaluate(const Document& pDocument) const {
         checkArgCount(2);
-        intrusive_ptr<const Value> pLeft(vpOperand[0]->evaluate(pDocument));
-        intrusive_ptr<const Value> pRight(vpOperand[1]->evaluate(pDocument));
+        Value pLeft(vpOperand[0]->evaluate(pDocument));
+        Value pRight(vpOperand[1]->evaluate(pDocument));
 
-        BSONType leftType = pLeft->getType();
-        BSONType rightType = pRight->getType();
+        BSONType leftType = pLeft.getType();
+        BSONType rightType = pRight.getType();
 
         uassert(16374, "$mod does not support dates", leftType != Date && rightType != Date);
 
@@ -1788,31 +1752,31 @@ namespace mongo {
         if (rightType == jstNULL || rightType == Undefined)
             return pRight;
         // ensure we aren't modding by 0
-        double right = pRight->coerceToDouble();
+        double right = pRight.coerceToDouble();
         if (right == 0)
-            return Value::getUndefined();
+            return Value(Undefined);
 
         if (leftType == NumberDouble) {
             // left is a double, return a double
-            double left = pLeft->coerceToDouble();
+            double left = pLeft.coerceToDouble();
             return Value::createDouble(fmod(left, right));
         }
-        else if (rightType == NumberDouble && pRight->coerceToInt() != right) {
+        else if (rightType == NumberDouble && pRight.coerceToInt() != right) {
             // the shell converts ints to doubles so if right is larger than int max or
             // if right truncates to something other than itself, it is a real double.
             // Integer-valued double case is handled below
-            double left = pLeft->coerceToDouble();
+            double left = pLeft.coerceToDouble();
             return Value::createDouble(fmod(left, right));
         }
         if (leftType == NumberLong || rightType == NumberLong) {
             // if either is long, return long
-            long long left = pLeft->coerceToLong();
-            long long rightLong = pRight->coerceToLong();
+            long long left = pLeft.coerceToLong();
+            long long rightLong = pRight.coerceToLong();
             return Value::createLong(left % rightLong);
         }
         // lastly they must both be ints, return int
-        int left = pLeft->coerceToInt();
-        int rightInt = pRight->coerceToInt();
+        int left = pLeft.coerceToInt();
+        int rightInt = pRight.coerceToInt();
         return Value::createInt(left % rightInt);
     }
 
@@ -1839,11 +1803,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionMonth::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionMonth::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_mon + 1); // MySQL uses 1-12 tm uses 0-11
     }
 
@@ -1865,8 +1828,7 @@ namespace mongo {
         ExpressionNary() {
     }
 
-    intrusive_ptr<const Value> ExpressionMultiply::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionMultiply::evaluate(const Document& pDocument) const {
         /*
           We'll try to return the narrowest possible result value.  To do that
           without creating intermediate Values, do the arithmetic for double
@@ -1879,13 +1841,13 @@ namespace mongo {
 
         const size_t n = vpOperand.size();
         for(size_t i = 0; i < n; ++i) {
-            intrusive_ptr<const Value> pValue(vpOperand[i]->evaluate(pDocument));
+            Value pValue(vpOperand[i]->evaluate(pDocument));
 
-            uassert(16375, "$multiply does not support dates", pValue->getType() != Date);
+            uassert(16375, "$multiply does not support dates", pValue.getType() != Date);
 
-            productType = Value::getWidestNumeric(productType, pValue->getType());
-            doubleProduct *= pValue->coerceToDouble();
-            longProduct *= pValue->coerceToLong();
+            productType = Value::getWidestNumeric(productType, pValue.getType());
+            doubleProduct *= pValue.coerceToDouble();
+            longProduct *= pValue.coerceToLong();
         }
 
         if (productType == NumberDouble)
@@ -1925,11 +1887,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionHour::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionHour::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_hour);
     }
 
@@ -1957,16 +1918,15 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionIfNull::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionIfNull::evaluate(const Document& pDocument) const {
         checkArgCount(2);
-        intrusive_ptr<const Value> pLeft(vpOperand[0]->evaluate(pDocument));
-        BSONType leftType = pLeft->getType();
+        Value pLeft(vpOperand[0]->evaluate(pDocument));
+        BSONType leftType = pLeft.getType();
 
         if ((leftType != Undefined) && (leftType != jstNULL))
             return pLeft;
 
-        intrusive_ptr<const Value> pRight(vpOperand[1]->evaluate(pDocument));
+        Value pRight(vpOperand[1]->evaluate(pDocument));
         return pRight;
     }
 
@@ -1995,7 +1955,7 @@ namespace mongo {
                 dynamic_cast<ExpressionConstant *>(pNew.get());
             if (pConst) {
                 ++constCount;
-                if (pConst->getValue()->getType() == String)
+                if (pConst->getValue().getType() == String)
                     ++stringCount;
             }
         }
@@ -2007,8 +1967,7 @@ namespace mongo {
           ExpressionConstant never refers to the argument Document.
         */
         if (constCount == n) {
-            intrusive_ptr<const Value> pResult(
-                evaluate(intrusive_ptr<Document>()));
+            Value pResult(evaluate(Document()));
             intrusive_ptr<Expression> pReplacement(
                 ExpressionConstant::create(pResult));
             return pReplacement;
@@ -2104,8 +2063,7 @@ namespace mongo {
               together before adding the result to the end of the expression
               operand vector.
             */
-            intrusive_ptr<const Value> pResult(
-                pConst->evaluate(intrusive_ptr<Document>()));
+            Value pResult(pConst->evaluate(Document()));
             pNew->addOperand(ExpressionConstant::create(pResult));
         }
 
@@ -2138,9 +2096,9 @@ namespace mongo {
         arrBuilder.doneFast();
     }
 
-    void ExpressionNary::addToBsonObj(
-        BSONObjBuilder *pBuilder, const std::string& fieldName,
-        bool requireExpression) const {
+    void ExpressionNary::addToBsonObj(BSONObjBuilder *pBuilder,
+                                      StringData fieldName,
+                                      bool requireExpression) const {
         BSONObjBuilder exprBuilder;
         toBson(&exprBuilder, getOpName());
         pBuilder->append(fieldName, exprBuilder.done());
@@ -2186,15 +2144,12 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionNot::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionNot::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pOp(vpOperand[0]->evaluate(pDocument));
+        Value pOp(vpOperand[0]->evaluate(pDocument));
 
-        bool b = pOp->coerceToBool();
-        if (b)
-            return Value::getFalse();
-        return Value::getTrue();
+        bool b = pOp.coerceToBool();
+        return Value(!b);
     }
 
     const char *ExpressionNot::getOpName() const {
@@ -2215,16 +2170,15 @@ namespace mongo {
         ExpressionNary() {
     }
 
-    intrusive_ptr<const Value> ExpressionOr::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionOr::evaluate(const Document& pDocument) const {
         const size_t n = vpOperand.size();
         for(size_t i = 0; i < n; ++i) {
-            intrusive_ptr<const Value> pValue(vpOperand[i]->evaluate(pDocument));
-            if (pValue->coerceToBool())
-                return Value::getTrue();
+            Value pValue(vpOperand[i]->evaluate(pDocument));
+            if (pValue.coerceToBool())
+                return Value(true);
         }
 
-        return Value::getFalse();
+        return Value(false);
     }
 
     void ExpressionOr::toMatcherBson(
@@ -2268,10 +2222,10 @@ namespace mongo {
           Evaluate and coerce the last argument to a boolean.  If it's true,
           then we can replace this entire expression.
          */
-        bool last = pLast->evaluate(intrusive_ptr<Document>())->coerceToBool();
+        bool last = pLast->evaluate(Document()).coerceToBool();
         if (last) {
             intrusive_ptr<ExpressionConstant> pFinal(
-                ExpressionConstant::create(Value::getTrue()));
+                ExpressionConstant::create(Value(true)));
             return pFinal;
         }
 
@@ -2317,11 +2271,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionSecond::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionSecond::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_sec);
     }
 
@@ -2349,22 +2302,22 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionStrcasecmp::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionStrcasecmp::evaluate(const Document& pDocument) const {
         checkArgCount(2);
-        intrusive_ptr<const Value> pString1(vpOperand[0]->evaluate(pDocument));
-        intrusive_ptr<const Value> pString2(vpOperand[1]->evaluate(pDocument));
+        Value pString1(vpOperand[0]->evaluate(pDocument));
+        Value pString2(vpOperand[1]->evaluate(pDocument));
 
         /* boost::iequals returns a bool not an int so strings must actually be allocated */
-        string str1 = boost::to_upper_copy( pString1->coerceToString() );
-        string str2 = boost::to_upper_copy( pString2->coerceToString() );
+        string str1 = boost::to_upper_copy( pString1.coerceToString() );
+        string str2 = boost::to_upper_copy( pString2.coerceToString() );
         int result = str1.compare(str2);
 
         if (result == 0)
-            return Value::getZero();
-        if (result > 0)
-            return Value::getOne();
-        return Value::getMinusOne();
+            return Value(0);
+        else if (result > 0)
+            return Value(1);
+        else
+            return Value(-1);
     }
 
     const char *ExpressionStrcasecmp::getOpName() const {
@@ -2391,28 +2344,27 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionSubstr::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionSubstr::evaluate(const Document& pDocument) const {
         checkArgCount(3);
-        intrusive_ptr<const Value> pString(vpOperand[0]->evaluate(pDocument));
-        intrusive_ptr<const Value> pLower(vpOperand[1]->evaluate(pDocument));
-        intrusive_ptr<const Value> pLength(vpOperand[2]->evaluate(pDocument));
+        Value pString(vpOperand[0]->evaluate(pDocument));
+        Value pLower(vpOperand[1]->evaluate(pDocument));
+        Value pLength(vpOperand[2]->evaluate(pDocument));
 
-        string str = pString->coerceToString();
+        string str = pString.coerceToString();
         uassert(16034, str::stream() << getOpName() <<
                 ":  starting index must be a numeric type (is BSON type " <<
-                typeName(pLower->getType()) << ")",
-                (pLower->getType() == NumberInt 
-                 || pLower->getType() == NumberLong 
-                 || pLower->getType() == NumberDouble));
+                typeName(pLower.getType()) << ")",
+                (pLower.getType() == NumberInt 
+                 || pLower.getType() == NumberLong 
+                 || pLower.getType() == NumberDouble));
         uassert(16035, str::stream() << getOpName() <<
                 ":  length must be a numeric type (is BSON type " <<
-                typeName(pLength->getType() )<< ")",
-                (pLength->getType() == NumberInt 
-                 || pLength->getType() == NumberLong 
-                 || pLength->getType() == NumberDouble));
-        string::size_type lower = static_cast< string::size_type >( pLower->coerceToLong() );
-        string::size_type length = static_cast< string::size_type >( pLength->coerceToLong() );
+                typeName(pLength.getType() )<< ")",
+                (pLength.getType() == NumberInt 
+                 || pLength.getType() == NumberLong 
+                 || pLength.getType() == NumberDouble));
+        string::size_type lower = static_cast< string::size_type >( pLower.coerceToLong() );
+        string::size_type length = static_cast< string::size_type >( pLength.coerceToLong() );
         if ( lower >= str.length() ) {
             // If lower > str.length() then string::substr() will throw out_of_range, so return an
             // empty string if lower is not a valid string index.
@@ -2445,27 +2397,26 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionSubtract::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionSubtract::evaluate(const Document& pDocument) const {
         BSONType productType;
         checkArgCount(2);
-        intrusive_ptr<const Value> pLeft(vpOperand[0]->evaluate(pDocument));
-        intrusive_ptr<const Value> pRight(vpOperand[1]->evaluate(pDocument));
+        Value pLeft(vpOperand[0]->evaluate(pDocument));
+        Value pRight(vpOperand[1]->evaluate(pDocument));
             
-        productType = Value::getWidestNumeric(pRight->getType(), pLeft->getType());
+        productType = Value::getWidestNumeric(pRight.getType(), pLeft.getType());
         
         uassert(16376,
                 "$subtract does not support dates",
-                pLeft->getType() != Date && pRight->getType() != Date);
+                pLeft.getType() != Date && pRight.getType() != Date);
 
         if (productType == NumberDouble) {
-            double right = pRight->coerceToDouble();
-            double left = pLeft->coerceToDouble();
+            double right = pRight.coerceToDouble();
+            double left = pLeft.coerceToDouble();
             return Value::createDouble(left - right);
         } 
 
-        long long right = pRight->coerceToLong();
-        long long left = pLeft->coerceToLong();
+        long long right = pRight.coerceToLong();
+        long long left = pLeft.coerceToLong();
         if (productType == NumberLong)
             return Value::createLong(left - right);
         else if (productType == NumberInt)
@@ -2498,11 +2449,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionToLower::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionToLower::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pString(vpOperand[0]->evaluate(pDocument));
-        string str = pString->coerceToString();
+        Value pString(vpOperand[0]->evaluate(pDocument));
+        string str = pString.coerceToString();
         boost::to_lower(str);
         return Value::createString(str);
     }
@@ -2531,11 +2481,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionToUpper::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionToUpper::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pString(vpOperand[0]->evaluate(pDocument));
-        string str(pString->coerceToString());
+        Value pString(vpOperand[0]->evaluate(pDocument));
+        string str(pString.coerceToString());
         boost::to_upper(str);
         return Value::createString(str);
     }
@@ -2563,11 +2512,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionWeek::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionWeek::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         int dayOfWeek = date.tm_wday;
         int dayOfYear = date.tm_yday;
         int prevSundayDayOfYear = dayOfYear - dayOfWeek; // may be negative
@@ -2611,11 +2559,10 @@ namespace mongo {
         ExpressionNary::addOperand(pExpression);
     }
 
-    intrusive_ptr<const Value> ExpressionYear::evaluate(
-        const intrusive_ptr<Document> &pDocument) const {
+    Value ExpressionYear::evaluate(const Document& pDocument) const {
         checkArgCount(1);
-        intrusive_ptr<const Value> pDate(vpOperand[0]->evaluate(pDocument));
-        tm date = pDate->coerceToTm();
+        Value pDate(vpOperand[0]->evaluate(pDocument));
+        tm date = pDate.coerceToTm();
         return Value::createInt(date.tm_year + 1900); // tm_year is years since 1900
     }
 
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index c9f13990854..e3bbf57a237 100755
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -29,6 +29,7 @@ namespace mongo {
     class BSONObjBuilder;
     class Builder;
     class Document;
+    class MutableDocument;
     class DocumentSource;
     class ExpressionContext;
     class Value;
@@ -76,8 +77,7 @@ namespace mongo {
 
           @returns the computed value
         */
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const = 0;
+        virtual Value evaluate(const Document& pDocument) const = 0;
 
         /*
           Add the Expression (and any descendant Expressions) into a BSON
@@ -94,9 +94,9 @@ namespace mongo {
             $project which distinguish between field inclusion and virtual
             field specification;  See ExpressionConstant.
          */
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const = 0;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const = 0;
 
         /*
           Add the Expression (and any descendant Expressions) into a BSON
@@ -223,9 +223,9 @@ namespace mongo {
     public:
         // virtuals from Expression
         virtual intrusive_ptr<Expression> optimize();
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const;
         virtual void addToBsonArray(BSONArrayBuilder *pBuilder) const;
         virtual void addDependencies(set<string>& deps, vector<string>* path=NULL) const;
 
@@ -313,7 +313,7 @@ namespace mongo {
     public:
         // virtuals from Expression
         virtual ~ExpressionAdd();
-        virtual intrusive_ptr<const Value> evaluate(const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
 
         // virtuals from ExpressionNary
@@ -334,8 +334,7 @@ namespace mongo {
         // virtuals from Expression
         virtual ~ExpressionAnd();
         virtual intrusive_ptr<Expression> optimize();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void toMatcherBson(BSONObjBuilder *pBuilder) const;
 
@@ -365,11 +364,10 @@ namespace mongo {
         virtual ~ExpressionCoerceToBool();
         virtual intrusive_ptr<Expression> optimize();
         virtual void addDependencies(set<string>& deps, vector<string>* path=NULL) const;
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const;
         virtual void addToBsonArray(BSONArrayBuilder *pBuilder) const;
 
         static intrusive_ptr<ExpressionCoerceToBool> create(
@@ -388,8 +386,7 @@ namespace mongo {
         // virtuals from ExpressionNary
         virtual ~ExpressionCompare();
         virtual intrusive_ptr<Expression> optimize();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -422,8 +419,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionCond();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -441,31 +437,29 @@ namespace mongo {
         virtual ~ExpressionConstant();
         virtual intrusive_ptr<Expression> optimize();
         virtual void addDependencies(set<string>& deps, vector<string>* path=NULL) const;
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const;
         virtual void addToBsonArray(BSONArrayBuilder *pBuilder) const;
 
         static intrusive_ptr<ExpressionConstant> createFromBsonElement(
             BSONElement *pBsonElement);
-        static intrusive_ptr<ExpressionConstant> create(
-            const intrusive_ptr<const Value> &pValue);
+        static intrusive_ptr<ExpressionConstant> create(const Value& pValue);
 
         /*
           Get the constant value represented by this Expression.
 
           @returns the value
          */
-        intrusive_ptr<const Value> getValue() const;
+        Value getValue() const;
 
     private:
         ExpressionConstant(BSONElement *pBsonElement);
-        ExpressionConstant(const intrusive_ptr<const Value> &pValue);
+        ExpressionConstant(const Value& pValue);
 
-        intrusive_ptr<const Value> pValue;
+        Value pValue;
     };
 
 
@@ -474,8 +468,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionDayOfMonth();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -491,8 +484,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionDayOfWeek();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -508,8 +500,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionDayOfYear();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -525,8 +516,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionDivide();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -544,11 +534,10 @@ namespace mongo {
         virtual ~ExpressionFieldPath();
         virtual intrusive_ptr<Expression> optimize();
         virtual void addDependencies(set<string>& deps, vector<string>* path=NULL) const;
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const;
         virtual void addToBsonArray(BSONArrayBuilder *pBuilder) const;
 
         /*
@@ -599,9 +588,9 @@ namespace mongo {
           @param pDocument current document traversed to (not the top-level one)
           @returns the field found; could be an array
          */
-        intrusive_ptr<const Value> evaluatePath(
+        Value evaluatePath(
             size_t index, const size_t pathLength, 
-            intrusive_ptr<Document> pDocument) const;
+            Document pDocument) const;
 
         FieldPath fieldPath;
     };
@@ -614,11 +603,10 @@ namespace mongo {
         virtual ~ExpressionFieldRange();
         virtual intrusive_ptr<Expression> optimize();
         virtual void addDependencies(set<string>& deps, vector<string>* path=NULL) const;
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const;
         virtual void addToBsonArray(BSONArrayBuilder *pBuilder) const;
         virtual void toMatcherBson(BSONObjBuilder *pBuilder) const;
 
@@ -644,7 +632,7 @@ namespace mongo {
          */
         static intrusive_ptr<ExpressionFieldRange> create(
             const intrusive_ptr<ExpressionFieldPath> &pFieldPath,
-            CmpOp cmpOp, const intrusive_ptr<const Value> &pValue);
+            CmpOp cmpOp, const Value& pValue);
 
         /*
           Add an intersecting range.
@@ -659,30 +647,30 @@ namespace mongo {
           @param cmpOp the comparison operator
           @param pValue the value to compare against
          */
-        void intersect(CmpOp cmpOp, const intrusive_ptr<const Value> &pValue);
+        void intersect(CmpOp cmpOp, const Value& pValue);
 
     private:
         ExpressionFieldRange(const intrusive_ptr<ExpressionFieldPath> &pFieldPath,
                              CmpOp cmpOp,
-                             const intrusive_ptr<const Value> &pValue);
+                             const Value& pValue);
 
         intrusive_ptr<ExpressionFieldPath> pFieldPath;
 
         class Range {
         public:
-            Range(CmpOp cmpOp, const intrusive_ptr<const Value> &pValue);
+            Range(CmpOp cmpOp, const Value& pValue);
             Range(const Range &rRange);
 
             Range *intersect(const Range *pRange) const;
-            bool contains(const intrusive_ptr<const Value> &pValue) const;
+            bool contains(const Value& pValue) const;
 
-            Range(const intrusive_ptr<const Value> &pBottom, bool bottomOpen,
-                  const intrusive_ptr<const Value> &pTop, bool topOpen);
+            Range(const Value& pBottom, bool bottomOpen,
+                  const Value& pTop, bool topOpen);
 
             bool bottomOpen;
             bool topOpen;
-            intrusive_ptr<const Value> pBottom;
-            intrusive_ptr<const Value> pTop;
+            Value pBottom;
+            Value pTop;
         };
 
         scoped_ptr<Range> pRange;
@@ -705,8 +693,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionHour();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -722,8 +709,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionIfNull();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -739,8 +725,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionMinute();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -756,8 +741,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionMod();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -773,8 +757,7 @@ namespace mongo {
     public:
         // virtuals from Expression
         virtual ~ExpressionMultiply();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
 
         // virtuals from ExpressionNary
@@ -797,8 +780,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionMonth();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -814,8 +796,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionNot();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -835,11 +816,10 @@ namespace mongo {
         virtual bool isSimple();
         virtual void addDependencies(set<string>& deps, vector<string>* path=NULL) const;
         /** Only evaluates non inclusion expressions.  For inclusions, use addToDocument(). */
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
-        virtual void addToBsonObj(
-            BSONObjBuilder *pBuilder, const std::string& fieldName,
-            bool requireExpression) const;
+        virtual Value evaluate(const Document& pDocument) const;
+        virtual void addToBsonObj(BSONObjBuilder *pBuilder,
+                                  StringData fieldName,
+                                  bool requireExpression) const;
         virtual void addToBsonArray(BSONArrayBuilder *pBuilder) const;
 
         /*
@@ -849,8 +829,7 @@ namespace mongo {
           @param pDocument the input Document
           @returns the result document
          */
-        intrusive_ptr<Document> evaluateDocument(
-            const intrusive_ptr<Document> &pDocument) const;
+        Document evaluateDocument(const Document& pDocument) const;
 
         /*
           evaluate(), but add the evaluated fields to a given document
@@ -860,9 +839,9 @@ namespace mongo {
           @param pDocument the input Document for this level
           @param rootDoc the root of the whole input document
          */
-        void addToDocument(const intrusive_ptr<Document>& pResult,
-                           const intrusive_ptr<Document>& pDocument,
-                           const intrusive_ptr<Document>& rootDoc
+        void addToDocument(MutableDocument& pResult,
+                           const Document& pDocument,
+                           const Document& rootDoc
                           ) const;
 
         // estimated number of fields that will be output
@@ -991,8 +970,7 @@ namespace mongo {
         // virtuals from Expression
         virtual ~ExpressionOr();
         virtual intrusive_ptr<Expression> optimize();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void toMatcherBson(BSONObjBuilder *pBuilder) const;
 
@@ -1020,8 +998,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionSecond();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1037,8 +1014,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionStrcasecmp();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1054,8 +1030,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionSubstr();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1071,8 +1046,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionSubtract();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1088,8 +1062,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionToLower();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1105,8 +1078,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionToUpper();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1122,8 +1094,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionWeek();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1139,8 +1110,7 @@ namespace mongo {
     public:
         // virtuals from ExpressionNary
         virtual ~ExpressionYear();
-        virtual intrusive_ptr<const Value> evaluate(
-            const intrusive_ptr<Document> &pDocument) const;
+        virtual Value evaluate(const Document& pDocument) const;
         virtual const char *getOpName() const;
         virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
 
@@ -1164,7 +1134,7 @@ namespace mongo {
         return 0;
     }
 
-    inline intrusive_ptr<const Value> ExpressionConstant::getValue() const {
+    inline Value ExpressionConstant::getValue() const {
         return pValue;
     }
 
diff --git a/src/mongo/db/pipeline/pipeline.cpp b/src/mongo/db/pipeline/pipeline.cpp
index b7ebf86272c..8616e688aac 100644
--- a/src/mongo/db/pipeline/pipeline.cpp
+++ b/src/mongo/db/pipeline/pipeline.cpp
@@ -416,7 +416,7 @@ namespace mongo {
             // cant use subArrayStart() due to error handling
             BSONArrayBuilder resultArray;
             for(bool hasDoc = !pSource->eof(); hasDoc; hasDoc = pSource->advance()) {
-                intrusive_ptr<Document> pDocument(pSource->getCurrent());
+                Document pDocument(pSource->getCurrent());
 
                 /* add the document to the result set */
                 BSONObjBuilder documentBuilder (resultArray.subobjStart());
@@ -476,8 +476,7 @@ namespace mongo {
         BSONArrayBuilder shardOpArray; // where we'll put the pipeline ops
         for(bool hasDocument = !pSourceBsonArray->eof(); hasDocument;
             hasDocument = pSourceBsonArray->advance()) {
-            intrusive_ptr<Document> pDocument(
-                pSourceBsonArray->getCurrent());
+            Document pDocument = pSourceBsonArray->getCurrent();
             BSONObjBuilder opBuilder;
             pDocument->toBson(&opBuilder);
             shardOpArray.append(opBuilder.obj());
diff --git a/src/mongo/db/pipeline/value.cpp b/src/mongo/db/pipeline/value.cpp
index 5ab0bd65c61..a6d6e85f878 100644
--- a/src/mongo/db/pipeline/value.cpp
+++ b/src/mongo/db/pipeline/value.cpp
@@ -14,157 +14,168 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "pch.h"
-#include "db/pipeline/value.h"
+#include "mongo/pch.h"
+
+#include "mongo/db/pipeline/value.h"
 
 #include <boost/functional/hash.hpp>
-#include "db/jsobj.h"
-#include "db/pipeline/builder.h"
-#include "db/pipeline/document.h"
-#include "util/mongoutils/str.h"
+
+#include "mongo/db/jsobj.h"
+#include "mongo/db/pipeline/builder.h"
+#include "mongo/db/pipeline/document.h"
+#include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
     using namespace mongoutils;
 
-    const intrusive_ptr<const Value> Value::pFieldUndefined(
-        new ValueStatic(Undefined));
-    const intrusive_ptr<const Value> Value::pFieldNull(new ValueStatic());
-    const intrusive_ptr<const Value> Value::pFieldTrue(new ValueStatic(true));
-    const intrusive_ptr<const Value> Value::pFieldFalse(new ValueStatic(false));
-    const intrusive_ptr<const Value> Value::pFieldMinusOne(new ValueStatic(-1));
-    const intrusive_ptr<const Value> Value::pFieldZero(new ValueStatic(0));
-    const intrusive_ptr<const Value> Value::pFieldOne(new ValueStatic(1));
+    void ValueStorage::putString(StringData s) {
+        const size_t sizeWithNUL = s.size() + 1;
+        if (sizeWithNUL <= sizeof(shortStrStorage)) {
+            shortStr = true;
+            shortStrSize = s.size();
+            memcpy(shortStrStorage, s.data(), sizeWithNUL);
+        }
+        else {
+            intrusive_ptr<const RCString> rcs = RCString::create(s);
+            fassert(16492, rcs);
+            genericRCPtr = rcs.get();
+            intrusive_ptr_add_ref(genericRCPtr);
+            refCounter = true;
+        }
+    }
+
+    void ValueStorage::putDocument(const Document& d) {
+        genericRCPtr = d._storage.get();
+
+        if (genericRCPtr) { // NULL here means empty document
+            intrusive_ptr_add_ref(genericRCPtr);
+            refCounter = true;
+        }
+    }
+
+    void ValueStorage::putVector(const RCVector* vec) {
+        fassert(16485, vec);
 
-    Value::~Value() {
+        genericRCPtr = vec;
+        intrusive_ptr_add_ref(genericRCPtr);
+        refCounter = true;
     }
 
-    Value::Value(): type(jstNULL) {}
+    Document ValueStorage::getDocument() const {
+        if (!genericRCPtr)
+            return Document();
 
-    Value::Value(BSONType theType): type(theType) {
-        switch(type) {
+        dassert(typeid(*genericRCPtr) == typeid(const DocumentStorage));
+        const DocumentStorage* documentPtr = static_cast<const DocumentStorage*>(genericRCPtr);
+        return Document(documentPtr);
+    }
+
+    Value::Value(BSONType theType): _storage(theType) {
+        switch(getType()) {
         case Undefined:
         case jstNULL:
         case Object: // empty
-        case Array: // empty
             break;
 
+        case Array: // empty
+            _storage.putVector(new RCVector());
+
         case Bool:
-            boolValue = false;
+            _storage.boolValue = false;
             break;
 
         case NumberDouble:
-            doubleValue = 0;
+            _storage.doubleValue = 0;
             break;
 
         case NumberInt:
-            intValue = 0;
+            _storage.intValue = 0;
             break;
 
         case NumberLong:
-            longValue = 0;
+            _storage.longValue = 0;
             break;
 
         case Date:
-            dateValue = 0;
+            _storage.dateValue = 0;
             break;
 
         case Timestamp:
-            timestampValue = 0;
+            _storage.timestampValue = 0;
             break;
 
         default:
             // nothing else is allowed
             uassert(16001, str::stream() <<
-                    "can't create empty Value of type " << typeName(type), false);
+                    "can't create empty Value of type " << typeName(getType()), false);
             break;
         }
     }
 
-    Value::Value(bool value)
-        : type(Bool)
-        , boolValue(value)
-    {}
 
-    intrusive_ptr<const Value> Value::createFromBsonElement(
-        BSONElement *pBsonElement) {
-        switch (pBsonElement->type()) {
-            case Undefined:
-                return getUndefined();
-            case jstNULL:
-                return getNull();
-            case Bool:
-                if (pBsonElement->boolean())
-                    return getTrue();
-                else
-                    return getFalse();
-            default:
-                intrusive_ptr<const Value> pValue(new Value(pBsonElement));
-                return pValue;
-        }
+    Value Value::createFromBsonElement(const BSONElement* pBsonElement) {
+        return Value(*pBsonElement);
     }
 
-    Value::Value(BSONElement *pBsonElement):
-        type(pBsonElement->type()),
-        pDocumentValue(),
-        vpValue() {
-        switch(type) {
+    Value::Value(const BSONElement& elem) : _storage(elem.type()) {
+        switch(getType()) {
         case NumberDouble:
-            doubleValue = pBsonElement->Double();
+            _storage.doubleValue = elem.Double();
             break;
 
         case String:
-            stringValue = pBsonElement->str();
+            _storage.putString(StringData(elem.valuestr(), elem.valuestrsize()-1));
             break;
 
         case Object: {
-            BSONObj document(pBsonElement->embeddedObject());
-            pDocumentValue = Document::createFromBsonObj(&document);
+            _storage.putDocument(Document(elem.embeddedObject()));
             break;
         }
 
         case Array: {
-            vector<BSONElement> vElement(pBsonElement->Array());
+            vector<BSONElement> vElement(elem.Array());
             const size_t n = vElement.size();
 
-            vpValue.reserve(n); // save on realloc()ing
+            intrusive_ptr<RCVector> vec (new RCVector);
+            vec->vec.reserve(n); // save on realloc()ing
 
             for(size_t i = 0; i < n; ++i) {
-                vpValue.push_back(
-                    Value::createFromBsonElement(&vElement[i]));
+                vec->vec.push_back(Value::createFromBsonElement(&vElement[i]));
             }
+            _storage.putVector(vec.get());
             break;
         }
 
         case jstOID:
-            BOOST_STATIC_ASSERT(sizeof(oidValue) == sizeof(OID));
-            memcpy(oidValue, pBsonElement->OID().getData(), sizeof(oidValue));
+            BOOST_STATIC_ASSERT(sizeof(_storage.oid) == sizeof(OID));
+            memcpy(_storage.oid, elem.OID().getData(), sizeof(OID));
             break;
 
         case Bool:
-            boolValue = pBsonElement->Bool();
+            _storage.boolValue = elem.boolean();
             break;
 
         case Date:
             // this is really signed but typed as unsigned for historical reasons
-            dateValue = static_cast<long long>(pBsonElement->Date().millis);
+            _storage.dateValue = static_cast<long long>(elem.date().millis);
             break;
 
         case RegEx:
-            stringValue = pBsonElement->regex();
-            // TODO pBsonElement->regexFlags();
+            _storage.putString(elem.regex());
+            // TODO elem.regexFlags();
             break;
 
         case NumberInt:
-            intValue = pBsonElement->numberInt();
+            _storage.intValue = elem.numberInt();
             break;
 
         case Timestamp:
             // asDate is a poorly named function that returns a ReplTime
-            timestampValue = pBsonElement->_opTime().asDate();
+            _storage.timestampValue = elem._opTime().asDate();
             break;
 
         case NumberLong:
-            longValue = pBsonElement->numberLong();
+            _storage.longValue = elem.numberLong();
             break;
 
         case Undefined:
@@ -182,210 +193,70 @@ namespace mongo {
         case Code:
         case MaxKey:
             uassert(16002, str::stream() <<
-                    "can't create Value of BSON type " << typeName(type), false);
+                    "can't create Value of BSON type " << typeName(getType()), false);
             break;
         }
     }
 
-    Value::Value(int value)
-        : type(NumberInt)
-        , intValue(value)
-    {}
-
-    intrusive_ptr<const Value> Value::createInt(int value) {
-        intrusive_ptr<const Value> pValue(new Value(value));
-        return pValue;
-    }
-
-    intrusive_ptr<const Value> Value::createIntOrLong(long long value) {
+    Value Value::createIntOrLong(long long value) {
         if (value > numeric_limits<int>::max() || value < numeric_limits<int>::min()) {
             // it is too large to be an int and should remain a long
-            return new Value(value);
+            return Value(value);
         }
 
         // should be an int since all arguments were int and it fits
         return createInt(value);
     }
 
-    Value::Value(long long value)
-        : type(NumberLong)
-        , longValue(value)
-    {}
-
-    intrusive_ptr<const Value> Value::createLong(long long value) {
-        intrusive_ptr<const Value> pValue(new Value(value));
-        return pValue;
-    }
-
-    Value::Value(double value)
-        : type(NumberDouble)
-        , doubleValue(value)
-    {}
-
-    intrusive_ptr<const Value> Value::createDouble(double value) {
-        intrusive_ptr<const Value> pValue(new Value(value));
-        return pValue;
-    }
-
-    intrusive_ptr<const Value> Value::createDate(const long long &value) {
+    Value Value::createDate(const long long value) {
         // Can't directly construct because constructor would clash with createLong
-        intrusive_ptr<Value> pValue(new Value(Date));
-        pValue->dateValue = value;
-        return pValue;
-    }
-
-    Value::Value(const OpTime& value)
-        : type(Timestamp)
-        , timestampValue(value.asDate())
-    {}
-
-    intrusive_ptr<const Value> Value::createTimestamp(const OpTime& value) {
-        intrusive_ptr<const Value> pValue(new Value(value));
-        return pValue;
-    }
-
-    Value::Value(const string &value):
-        type(String),
-        pDocumentValue(),
-        vpValue() {
-        stringValue = value;
-    }
-
-    intrusive_ptr<const Value> Value::createString(const string &value) {
-        intrusive_ptr<const Value> pValue(new Value(value));
-        return pValue;
-    }
-
-    Value::Value(const intrusive_ptr<Document> &pDocument):
-        type(Object),
-        pDocumentValue(pDocument),
-        vpValue() {
-    }
-
-    intrusive_ptr<const Value> Value::createDocument(
-        const intrusive_ptr<Document> &pDocument) {
-        intrusive_ptr<const Value> pValue(new Value(pDocument));
-        return pValue;
-    }
-
-    Value::Value(const vector<intrusive_ptr<const Value> > &thevpValue):
-        type(Array),
-        pDocumentValue(),
-        vpValue(thevpValue) {
-    }
-
-    intrusive_ptr<const Value> Value::createArray(
-        const vector<intrusive_ptr<const Value> > &vpValue) {
-        intrusive_ptr<const Value> pValue(new Value(vpValue));
-        return pValue;
+        Value val (Date);
+        val._storage.dateValue = value;
+        return val;
     }
 
     double Value::getDouble() const {
         BSONType type = getType();
         if (type == NumberInt)
-            return intValue;
+            return _storage.intValue;
         if (type == NumberLong)
-            return static_cast< double >( longValue );
+            return static_cast< double >( _storage.longValue );
 
         verify(type == NumberDouble);
-        return doubleValue;
-    }
-
-    string Value::getString() const {
-        verify(getType() == String);
-        return stringValue;
+        return _storage.doubleValue;
     }
 
-    intrusive_ptr<Document> Value::getDocument() const {
+    Document Value::getDocument() const {
         verify(getType() == Object);
-        return pDocumentValue;
-    }
-
-    ValueIterator::~ValueIterator() {
-    }
-
-    Value::vi::~vi() {
-    }
-
-    bool Value::vi::more() const {
-        return (nextIndex < size);
-    }
-
-    intrusive_ptr<const Value> Value::vi::next() {
-        verify(more());
-        return (*pvpValue)[nextIndex++];
-    }
-
-    Value::vi::vi(const intrusive_ptr<const Value> &pValue,
-                  const vector<intrusive_ptr<const Value> > *thepvpValue):
-        size(thepvpValue->size()),
-        nextIndex(0),
-        pvpValue(thepvpValue) {
-    }
-
-    intrusive_ptr<ValueIterator> Value::getArray() const {
-        verify(getType() == Array);
-        intrusive_ptr<ValueIterator> pVI(
-            new vi(intrusive_ptr<const Value>(this), &vpValue));
-        return pVI;
-    }
-
-    OID Value::getOid() const {
-        verify(getType() == jstOID);
-        return OID(oidValue);
-    }
-
-    bool Value::getBool() const {
-        verify(getType() == Bool);
-        return boolValue;
+        return _storage.getDocument();
     }
 
-    long long Value::getDate() const {
-        verify(getType() == Date);
-        return dateValue;
-    }
+    Value Value::operator[] (size_t index) const {
+        if (missing() || getType() != Array || index >= getArrayLength())
+            return Value();
 
-    OpTime Value::getTimestamp() const {
-        verify(getType() == Timestamp);
-        return timestampValue;
+        return getArray()[index];
     }
 
-    string Value::getRegex() const {
-        verify(getType() == RegEx);
-        return stringValue;
-    }
+    Value Value::operator[] (StringData name) const {
+        if (missing() || getType() != Object)
+            return Value();
 
-    string Value::getSymbol() const {
-        verify(getType() == Symbol);
-        return stringValue;
+        return getDocument()[name];
     }
 
-    int Value::getInt() const {
-        verify(getType() == NumberInt);
-        return intValue;
-    }
-
-    long long Value::getLong() const {
-        BSONType type = getType();
-        if (type == NumberInt)
-            return intValue;
-
-        verify(type == NumberLong);
-        return longValue;
-    }
-
-    void Value::addToBson(Builder *pBuilder) const {
+    void Value::addToBson(Builder* pBuilder) const {
         switch(getType()) {
         case NumberDouble:
             pBuilder->append(getDouble());
             break;
 
         case String:
-            pBuilder->append(getString());
+            pBuilder->append(getStringData());
             break;
 
         case Object: {
-            intrusive_ptr<Document> pDocument(getDocument());
+            Document pDocument(getDocument());
             BSONObjBuilder subBuilder;
             pDocument->toBson(&subBuilder);
             subBuilder.done();
@@ -394,10 +265,10 @@ namespace mongo {
         }
 
         case Array: {
-            const size_t n = vpValue.size();
+            const size_t n = getArray().size();
             BSONArrayBuilder arrayBuilder(n);
             for(size_t i = 0; i < n; ++i) {
-                vpValue[i]->addToBsonArray(&arrayBuilder);
+                getArray()[i].addToBsonArray(&arrayBuilder);
             }
 
             pBuilder->append(&arrayBuilder);
@@ -464,25 +335,23 @@ namespace mongo {
         }
     }
 
-    void Value::addToBsonObj(BSONObjBuilder *pBuilder, const std::string& fieldName) const {
+    void Value::addToBsonObj(BSONObjBuilder* pBuilder, StringData fieldName) const {
+        if (missing()) return;
+
         BuilderObj objBuilder(pBuilder, fieldName);
         addToBson(&objBuilder);
     }
 
-    void Value::addToBsonArray(BSONArrayBuilder *pBuilder) const {
+    void Value::addToBsonArray(BSONArrayBuilder* pBuilder) const {
+        if (missing()) return;
+
         BuilderArray arrBuilder(pBuilder);
         addToBson(&arrBuilder);
     }
 
     bool Value::coerceToBool() const {
         // TODO Unify the implementation with BSONElement::trueValue().
-        BSONType type = getType();
-        switch(type) {
-        case NumberDouble:
-            if (doubleValue != 0)
-                return true;
-            break;
-
+        switch(getType()) {
         case String:
         case Object:
         case Array:
@@ -494,135 +363,112 @@ namespace mongo {
         case Timestamp:
             return true;
 
-        case Bool:
-            if (boolValue)
-                return true;
-            break;
-
-        case CodeWScope:
-            verify(false); // CW TODO unimplemented
-            break;
-
-        case NumberInt:
-            if (intValue != 0)
-                return true;
-            break;
-
-        case NumberLong:
-            if (longValue != 0)
-                return true;
-            break;
-
         case jstNULL:
         case Undefined:
-            /* nothing to do */
-            break;
+            return false;
+
+        case Bool: return _storage.boolValue;
+        case NumberInt: return _storage.intValue;
+        case NumberLong: return _storage.longValue;
+        case NumberDouble: return _storage.doubleValue;
 
             /* these shouldn't happen in this context */
+        case CodeWScope:
         case MinKey:
         case EOO:
         case DBRef:
         case Code:
         case MaxKey:
+        default:
             verify(false); // CW TODO better message
-            break;
         }
-
-        return false;
     }
 
     int Value::coerceToInt() const {
-        switch(type) {
+        switch(getType()) {
         case NumberDouble:
-            return (int)doubleValue;
+            return _storage.doubleValue;
 
         case NumberInt:
-            return intValue;
+            return _storage.intValue;
 
         case NumberLong:
-            return (int)longValue;
+            return _storage.longValue;
 
         case jstNULL:
         case Undefined:
-            break;
+            return 0;
 
         case String:
         default:
             uassert(16003, str::stream() <<
-                    "can't convert from BSON type " << typeName(type) <<
+                    "can't convert from BSON type " << typeName(getType()) <<
                     " to int",
                     false);
-        } // switch(type)
-
-        return (int)0;
+        } // switch(getType())
     }
 
     long long Value::coerceToLong() const {
-        switch(type) {
+        switch(getType()) {
         case NumberDouble:
-            return (long long)doubleValue;
+            return _storage.doubleValue;
 
         case NumberInt:
-            return intValue;
+            return _storage.intValue;
 
         case NumberLong:
-            return longValue;
+            return _storage.longValue;
 
         case jstNULL:
         case Undefined:
-            break;
+            return 0;
 
         case String:
         default:
             uassert(16004, str::stream() <<
-                    "can't convert from BSON type " << typeName(type) <<
+                    "can't convert from BSON type " << typeName(getType()) <<
                     " to long",
                     false);
-        } // switch(type)
-
-        return (long long)0;
+        } // switch(getType())
     }
 
     double Value::coerceToDouble() const {
-        switch(type) {
+        switch(getType()) {
         case NumberDouble:
-            return doubleValue;
+            return _storage.doubleValue;
 
         case NumberInt:
-            return (double)intValue;
+            return _storage.intValue;
 
         case NumberLong:
-            return (double)longValue;
+            return _storage.longValue;
 
         case jstNULL:
         case Undefined:
-            break;
+            return 0;
 
         case String:
         default:
             uassert(16005, str::stream() <<
-                    "can't convert from BSON type " << typeName(type) <<
+                    "can't convert from BSON type " << typeName(getType()) <<
                     " to double",
                     false);
-        } // switch(type)
-
-        return (double)0;
+        } // switch(getType())
     }
 
     long long Value::coerceToDate() const {
-        switch(type) {
-
+        switch(getType()) {
         case Date:
-            return dateValue; 
+            return getDate();
 
         case Timestamp:
             return getTimestamp().getSecs() * 1000LL;
 
         default:
             uassert(16006, str::stream() <<
-                    "can't convert from BSON type " << typeName(type) << " to Date",
+                    "can't convert from BSON type " << typeName(getType()) << " to Date",
                     false);
-        } // switch(type)
+        } // switch(getType())
     }
 
     time_t Value::coerceToTimeT() const {
@@ -678,21 +524,21 @@ namespace mongo {
 
     string Value::coerceToString() const {
         stringstream ss;
-        switch(type) {
+        switch(getType()) {
         case NumberDouble:
-            ss << doubleValue;
+            ss << _storage.doubleValue;
             return ss.str();
 
         case NumberInt:
-            ss << intValue;
+            ss << _storage.intValue;
             return ss.str();
 
         case NumberLong:
-            ss << longValue;
+            ss << _storage.longValue;
             return ss.str();
 
         case String:
-            return stringValue;
+            return getString();
 
         case Timestamp:
             ss << getTimestamp().toStringPretty();
@@ -703,36 +549,33 @@ namespace mongo {
 
         case jstNULL:
         case Undefined:
-            break;
+            return "";
 
         default:
             uassert(16007, str::stream() <<
-                    "can't convert from BSON type " << typeName(type) <<
+                    "can't convert from BSON type " << typeName(getType()) <<
                     " to String",
                     false);
-        } // switch(type)
-
-        return "";
+        } // switch(getType())
     }
 
     OpTime Value::coerceToTimestamp() const {
-        switch(type) {
-
+        switch(getType()) {
         case Timestamp:
-            return timestampValue;
+            return getTimestamp();
 
         default:
             uassert(16378, str::stream() <<
-                    "can't convert from BSON type " << typeName(type) <<
+                    "can't convert from BSON type " << typeName(getType()) <<
                     " to timestamp",
                     false);
-        } // switch(type)
+        } // switch(getType())
     }
 
-    int Value::compare(const intrusive_ptr<const Value> &rL,
-                       const intrusive_ptr<const Value> &rR) {
-        BSONType lType = rL->getType();
-        BSONType rType = rR->getType();
+    int Value::compare(const Value& rL, const Value& rR) {
+        // missing is treated as undefined for compatibility with BSONObj::woCompare
+        BSONType lType = rL.missing() ? Undefined : rL.getType();
+        BSONType rType = rR.missing() ? Undefined : rR.getType();
 
         /*
           Special handling for Undefined and NULL values; these are types,
@@ -775,8 +618,8 @@ namespace mongo {
 
             /* if the biggest type of either is a double, compare as doubles */
             if ((lType == NumberDouble) || (rType == NumberDouble)) {
-                const double left = rL->getDouble();
-                const double right = rR->getDouble();
+                const double left = rL.getDouble();
+                const double right = rR.getDouble();
                 if (left < right)
                     return -1;
                 if (left > right)
@@ -786,8 +629,8 @@ namespace mongo {
 
             /* if the biggest type of either is a long, compare as longs */
             if ((lType == NumberLong) || (rType == NumberLong)) {
-                const long long left = rL->getLong();
-                const long long right = rR->getLong();
+                const long long left = rL.getLong();
+                const long long right = rR.getLong();
                 if (left < right)
                     return -1;
                 if (left > right)
@@ -797,8 +640,8 @@ namespace mongo {
 
             /* if we got here, they must both be ints; compare as ints */
             {
-                const int left = rL->getInt();
-                const int right = rR->getInt();
+                const int left = rL.getInt();
+                const int right = rR.getInt();
                 if (left < right)
                     return -1;
                 if (left > right)
@@ -820,40 +663,47 @@ namespace mongo {
             /* these types were handled above */
             verify(false);
 
-        case String:
-            return rL->stringValue.compare(rR->stringValue);
+        case String: {
+            StringData r = rR.getStringData();
+            StringData l = rL.getStringData();
+            size_t bytes = min(r.size(), l.size());
+            int ret = memcmp(l.data(), r.data(), bytes);
+            if (ret)
+                return ret;
+            return l.size() - r.size();
+        }
 
         case Object:
-            return Document::compare(rL->getDocument(), rR->getDocument());
+            return Document::compare(rL.getDocument(), rR.getDocument());
 
         case Array: {
-            intrusive_ptr<ValueIterator> pli(rL->getArray());
-            intrusive_ptr<ValueIterator> pri(rR->getArray());
-
-            while(true) {
-                /* have we run out of left array? */
-                if (!pli->more()) {
-                    if (!pri->more())
-                        return 0; // the arrays are the same length
-
-                    return -1; // the left array is shorter
-                }
-
-                /* have we run out of right array? */
-                if (!pri->more())
-                    return 1; // the right array is shorter
-
-                /* compare the two corresponding elements */
-                intrusive_ptr<const Value> plv(pli->next());
-                intrusive_ptr<const Value> prv(pri->next());
-                const int cmp = Value::compare(plv, prv);
+            const vector<Value>& lArr = rL.getArray();
+            const vector<Value>& rArr = rR.getArray();
+
+            vector<Value>::const_iterator lIt =  lArr.begin();
+            vector<Value>::const_iterator lEnd = lArr.end();
+            vector<Value>::const_iterator rIt =  rArr.begin();
+            vector<Value>::const_iterator rEnd = rArr.end();
+
+            for ( ; lIt != lEnd && rIt != rEnd; ++lIt, ++rIt ) {
+                // compare the two corresponding elements
+                const int cmp = Value::compare(*lIt, *rIt);
                 if (cmp)
                     return cmp; // values are unequal
             }
 
-            /* NOTREACHED */
-            verify(false);
-            break;
+            if (lIt == lEnd && rIt == rEnd) {
+                return 0; // the arrays are the same length
+            }
+            else if (lIt == lEnd && rIt != rEnd) {
+                return -1; // the left array is shorter
+            }
+            else if (lIt != lEnd && rIt == rEnd) {
+                return 1; // the right array is shorter
+            }
+            else {
+                verify(false); // we shouldn't have exited the loop in this case
+            }
         }
 
         case BinData:
@@ -862,26 +712,25 @@ namespace mongo {
             uassert(16017, str::stream() <<
                     "comparisons of values of BSON type " << typeName(lType) <<
                     " are not supported", false);
-            // pBuilder->appendBinData(fieldName, ...);
             break;
 
         case jstOID:
-            if (rL->getOid() < rR->getOid())
+            if (rL.getOid() < rR.getOid())
                 return -1;
-            if (rL->getOid() == rR->getOid())
+            if (rL.getOid() == rR.getOid())
                 return 0;
             return 1;
 
         case Bool:
-            if (rL->boolValue == rR->boolValue)
+            if (rL.getBool() == rR.getBool())
                 return 0;
-            if (rL->boolValue)
+            if (rL.getBool())
                 return 1;
             return -1;
 
         case Date: {
-            long long l = rL->dateValue;
-            long long r = rR->dateValue;
+            long long l = rL.getDate();
+            long long r = rR.getDate();
             if (l < r)
                 return -1;
             if (l > r)
@@ -890,12 +739,12 @@ namespace mongo {
         }
 
         case RegEx:
-            return rL->stringValue.compare(rR->stringValue);
+            return rL.getRegex().compare(rR.getRegex());
 
         case Timestamp:
-            if (rL->timestampValue < rR->timestampValue)
+            if (rL.getTimestamp() < rR.getTimestamp())
                 return -1;
-            if (rL->timestampValue > rR->timestampValue)
+            if (rL.getTimestamp() > rR.getTimestamp())
                 return 1;
             return 0;
 
@@ -910,17 +759,13 @@ namespace mongo {
         case Code:
         case MaxKey:
             verify(false);
-            break;
         } // switch(lType)
 
-        /* NOTREACHED */
-        return 0;
+        verify(false);
     }
 
     void Value::hash_combine(size_t &seed) const {
-        BSONType type = getType();
-
-        switch(type) {
+        switch(getType()) {
             /*
               Numbers whose values are equal need to hash to the same thing
               as well.  Note that Value::compare() promotes numeric values to
@@ -933,27 +778,25 @@ namespace mongo {
              */
         case NumberDouble:
         case NumberLong:
-        case NumberInt:
-        {
-            const double d = getDouble();
-            boost::hash_combine(seed, d);
+        case NumberInt: {
+            boost::hash_combine(seed, getDouble());
             break;
         }
 
-        case String:
-            boost::hash_combine(seed, stringValue);
+        case String: {
+            StringData sd = getStringData();
+            boost::hash_range(seed, sd.data(), (sd.data() + sd.size()));
             break;
+        }
 
         case Object:
             getDocument()->hash_combine(seed);
             break;
 
         case Array: {
-            intrusive_ptr<ValueIterator> pIter(getArray());
-            while(pIter->more()) {
-                intrusive_ptr<const Value> pValue(pIter->next());
-                pValue->hash_combine(seed);
-            };
+            const vector<Value>& vec = getArray();
+            for (size_t i=0; i < vec.size(); i++)
+                vec[i].hash_combine(seed);
             break;
         }
 
@@ -961,7 +804,7 @@ namespace mongo {
         case Symbol:
         case CodeWScope:
             uassert(16018, str::stream() <<
-                    "hashes of values of BSON type " << typeName(type) <<
+                    "hashes of values of BSON type " << typeName(getType()) <<
                     " are not supported", false);
             break;
 
@@ -970,19 +813,19 @@ namespace mongo {
             break;
 
         case Bool:
-            boost::hash_combine(seed, boolValue);
+            boost::hash_combine(seed, getBool());
             break;
 
         case Date:
-            boost::hash_combine(seed, dateValue);
+            boost::hash_combine(seed, getDate());
             break;
 
         case RegEx:
-            boost::hash_combine(seed, stringValue);
+            boost::hash_combine(seed, getRegex());
             break;
 
         case Timestamp:
-            boost::hash_combine(seed, timestampValue);
+            boost::hash_combine(seed, _storage.timestampValue);
             break;
 
         case Undefined:
@@ -997,7 +840,7 @@ namespace mongo {
         case MaxKey:
             verify(false); // CW TODO better message
             break;
-        } // switch(type)
+        } // switch(getType())
     }
 
     BSONType Value::getWidestNumeric(BSONType lType, BSONType rType) {
@@ -1067,18 +910,19 @@ namespace mongo {
     }
 
     size_t Value::getApproximateSize() const {
-        switch(type) {
+        switch(getType()) {
         case String:
-            return sizeof(Value) + stringValue.length();
+            return sizeof(Value) + sizeof(RCString) + getStringData().size();
 
         case Object:
-            return sizeof(Value) + pDocumentValue->getApproximateSize();
+            return sizeof(Value) + getDocument()->getApproximateSize();
 
         case Array: {
             size_t size = sizeof(Value);
-            const size_t n = vpValue.size();
+            size += sizeof(RCVector);
+            const size_t n = getArray().size();
             for(size_t i = 0; i < n; ++i) {
-                size += vpValue[i]->getApproximateSize();
+                size += getArray()[i].getApproximateSize();
             }
             return size;
         }
@@ -1105,7 +949,6 @@ namespace mongo {
         case Code:
         case MaxKey:
             verify(false); // CW TODO better message
-            return sizeof(Value);
         }
 
         /*
@@ -1116,14 +959,58 @@ namespace mongo {
           this final catch-all is here.
          */
         verify(false);
-        return sizeof(Value);
     }
 
+    string Value::toString() const {
+        // TODO use StringBuilder when operator << is ready
+        stringstream out;
+        out << *this;
+        return out.str();
+    }
+
+    ostream& operator << (ostream& out, const Value& val) {
+        if (val.missing()) return out << "MISSING";
+
+        switch(val.getType()) {
+        case jstOID: return out << val.getOid();
+        case String: return out << '"' << val.getString() << '"';
+        case RegEx: return out << '/' << val.getRegex() << '/';
+        case Symbol: return out << val.getSymbol();
+        case Bool: return out << (val.getBool() ? "true" : "false");
+        case NumberDouble: return out << val.getDouble();
+        case NumberLong: return out << val.getLong();
+        case NumberInt: return out << val.getInt();
+        case jstNULL: return out << "null";
+        case Undefined: return out << "undefined";
+        case Date: return out << Date_t(val.getDate()).toString();
+        case Timestamp: return out << val.getTimestamp().toString();
+        case Object: return out << val.getDocument()->toString();
+        case Array: {
+            out << "[";
+            const size_t n = val.getArray().size();
+            for(size_t i = 0; i < n; i++) {
+                if (i)
+                    out << ", ";
+                out << val.getArray()[i];
+            }
+            out << "]";
+            return out;
+        }
 
-    void ValueStatic::addRef() const {
-    }
+            /* these shouldn't happen in this context */
+        case CodeWScope:
+        case BinData: 
+        case MinKey:
+        case EOO:
+        case DBRef:
+        case Code:
+        case MaxKey:
+            verify(false); // CW TODO better message
+        }
 
-    void ValueStatic::release() const {
+
+        // Not in default case to trigger better warning if a case is missing
+        verify(false);
     }
 
 }
diff --git a/src/mongo/db/pipeline/value.h b/src/mongo/db/pipeline/value.h
index 6a99b673a70..fe16446e968 100755..100644
--- a/src/mongo/db/pipeline/value.h
+++ b/src/mongo/db/pipeline/value.h
@@ -16,465 +16,285 @@
 
 #pragma once
 
-#include "mongo/pch.h"
-#include "bson/bsontypes.h"
-#include "bson/oid.h"
-#include "util/intrusive_counter.h"
-#include "util/optime.h"
+#include "mongo/db/pipeline/value_internal.h"
 
 namespace mongo {
     class BSONElement;
     class Builder;
-    class Document;
-    class Value;
 
-    class ValueIterator :
-        public IntrusiveCounterUnsigned {
-    public:
-        virtual ~ValueIterator();
-
-        /*
-          Ask if there are more fields to return.
-
-          @returns true if there are more fields, false otherwise
-        */
-        virtual bool more() const = 0;
-
-        /*
-          Move the iterator to point to the next field and return it.
-
-          @returns the next field's <name, Value>
-        */
-        virtual intrusive_ptr<const Value> next() = 0;
-    };
-
-
-    /*
-      Values are immutable, so these are passed around as
-      intrusive_ptr<const Value>.
+    /** A variant type that can hold any type of data representable in BSON
+     *
+     *  Small values are stored inline, but some values, such as large strings,
+     *  are heap allocated. It has smart pointer capabilities built-in so it is
+     *  safe and recommended to pass these around and return them by value.
+     *
+     *  Values are immutable, but can be assigned. This means that once you have
+     *  a Value, you can be assured that none of the data in that Value will
+     *  change. However if you have a non-const Value you replace it with
+     *  operator=. These rules are the same as BSONObj, and similar to
+     *  shared_ptr<const Object> with stronger guarantees of constness. This is
+     *  also the same as Java's String type.
+     *
+     *  Thread-safety: A single Value instance can be safely shared between
+     *  threads as long as there are no writers while other threads are
+     *  accessing the object. Any number of threads can read from a Value
+     *  concurrently. There are no restrictions on how threads access Value
+     *  instances exclusively owned by them, even if they reference the same
+     *  storage as Value in other threads.
      */
-    class Value :
-        public IntrusiveCounterUnsigned {
+    class Value {
     public:
-        ~Value();
-
-        /*
-          Construct a Value from a BSONElement.
-
-          This ignores the name of the element, and only uses the value,
-          whatever type it is.
-
-          @returns a new Value initialized from the bsonElement
-        */
-        static intrusive_ptr<const Value> createFromBsonElement(
-            BSONElement *pBsonElement);
-
-        /*
-          Construct an integer-valued Value.
-
-          For commonly used values, consider using one of the singleton
-          instances defined below.
-
-          @param value the value
-          @returns a Value with the given value
-        */
-        static intrusive_ptr<const Value> createInt(int value);
-
-        /*
-          Construct a long or interger-valued Value.
-          Used when preforming arithmetic operations with int where the result may be too large
-          and need to be stored as long. The Value will be an int if value fits, otherwise it
-          will be a long.
-
-          @param value the value
-          @returns a Value with the given value
-        */
-        static intrusive_ptr<const Value> createIntOrLong(long long value);
-
-        /*
-          Construct an long(long)-valued Value.
-
-          For commonly used values, consider using one of the singleton
-          instances defined below.
-
-          @param value the value
-          @returns a Value with the given value
-        */
-        static intrusive_ptr<const Value> createLong(long long value);
-
-        /*
-          Construct a double-valued Value.
-
-          @param value the value
-          @returns a Value with the given value
-        */
-        static intrusive_ptr<const Value> createDouble(double value);
-
-        /*
-          Construct a string-valued Value.
-
-          @param value the value
-          @returns a Value with the given value
-        */
-        static intrusive_ptr<const Value> createString(const string &value);
-
-        /*
-          Construct a date-valued Value.
-
-          @param value the value
-          @returns a Value with the given value
-        */
-        static intrusive_ptr<const Value> createDate(const long long &value);
-
-        static intrusive_ptr<const Value> createTimestamp(const OpTime& value);
-
-        /*
-          Construct a document-valued Value.
+        /** Construct a Value
+         *
+         *  All types not listed will be rejected rather than converted (see private for why)
+         *
+         *  Note: Currently these are all explicit conversions.
+         *        I'm not sure if we want implicit or not.
+         *  //TODO decide
+         */
 
-          @param value the value
-          @returns a Value with the given value
-        */
-        static intrusive_ptr<const Value> createDocument(
-            const intrusive_ptr<Document> &pDocument);
+        Value(): _storage() {} // "Missing" value
+        explicit Value(bool value) : _storage(Bool, value) {}
+        explicit Value(int value) : _storage(NumberInt, value) {}
+        explicit Value(long long value) : _storage(NumberLong, value) {}
+        explicit Value(double value) : _storage(NumberDouble, value) {}
+        explicit Value(const OpTime& value) : _storage(Timestamp, value.asDate()) {}
+        explicit Value(StringData value) : _storage(String, value) {}
+        explicit Value(const string& value) : _storage(String, StringData(value)) {}
+        explicit Value(const char* value) : _storage(String, StringData(value)) {}
+        explicit Value(const Document& doc) : _storage(Object, doc) {}
+        explicit Value(const vector<Value>& vec) : _storage(Array, new RCVector(vec)) {}
+
+        /** Creates an empty or zero value of specified type.
+         *  This is currently the only way to create Undefined or Null Values.
+         */
+        explicit Value(BSONType type);
 
-        /*
-          Construct an array-valued Value.
+        // TODO: add an unsafe version that can share storage with the BSONElement
+        /// Deep-convert from BSONElement to Value
+        explicit Value(const BSONElement& elem);
 
-          @param value the value
-          @returns a Value with the given value
+        /** Construct a long or integer-valued Value.
+         *
+         *  Used when preforming arithmetic operations with int where the
+         *  result may be too large and need to be stored as long. The Value
+         *  will be an int if value fits, otherwise it will be a long.
         */
-        static intrusive_ptr<const Value> createArray(
-            const vector<intrusive_ptr<const Value> > &vpValue);
+        static Value createIntOrLong(long long value);
 
-        /*
-          Get the BSON type of the field.
-
-          If the type is jstNULL, no value getter will work.
-
-          @return the BSON type of the field.
-        */
-        BSONType getType() const;
+        /** A "missing" value indicates the lack of a Value.
+         *  This is similar to undefined/null but should not appear in output to BSON.
+         *  Missing Values are returned by Document when accessing non-existent fields.
+         */
+        bool missing() const { return _storage.type == EOO; }
 
-        /*
-          Getters.
+        /** Get the BSON type of the field.
+         *  Warning: currently asserts if missing. This will probably change in the future.
+         */
+        BSONType getType() const { return _storage.bsonType(); }
 
-          @returns the Value's value; asserts if the requested value type is
-          incorrect.
-        */
+        /** Exact type getters.
+         *  Asserts if the requested value type is not exactly correct.
+         *  See coerceTo methods below for a more type-flexible alternative.
+         */
         double getDouble() const;
         string getString() const;
-        intrusive_ptr<Document> getDocument() const;
-        intrusive_ptr<ValueIterator> getArray() const;
+        StringData getStringData() const; // May contain embedded NUL bytes
+        Document getDocument() const;
         OID getOid() const;
         bool getBool() const;
-        long long getDate() const;
+        long long getDate() const; // in milliseconds
         OpTime getTimestamp() const;
         string getRegex() const;
         string getSymbol() const;
         int getInt() const;
         long long getLong() const;
-
-        /*
-          Get the length of an array value.
-
-          @returns the length of the array, if this is array-valued; otherwise
-             throws an error
-        */
+        const vector<Value>& getArray() const { return _storage.getArray(); }
         size_t getArrayLength() const;
 
-        /*
-          Add this value to the BSON object under construction.
-        */
-        void addToBsonObj(BSONObjBuilder *pBuilder, const std::string& fieldName) const;
+        /// Access an element of a subarray. Returns Value() if missing or getType() != Array
+        Value operator[] (size_t index) const;
 
-        /*
-          Add this field to the BSON array under construction.
+        /// Access a field of a subdocument. Returns Value() if missing or getType() != Object
+        Value operator[] (StringData name) const;
 
-          As part of an array, the Value's name will be ignored.
-        */
-        void addToBsonArray(BSONArrayBuilder *pBuilder) const;
+        /// Add this value to the BSON object under construction.
+        void addToBsonObj(BSONObjBuilder* pBuilder, StringData fieldName) const;
 
-        /*
-          Get references to singleton instances of commonly used field values.
-         */
-        static intrusive_ptr<const Value> getUndefined();
-        static intrusive_ptr<const Value> getNull();
-        static intrusive_ptr<const Value> getTrue();
-        static intrusive_ptr<const Value> getFalse();
-        static intrusive_ptr<const Value> getMinusOne();
-        static intrusive_ptr<const Value> getZero();
-        static intrusive_ptr<const Value> getOne();
-
-        /**
-         * Coerce (cast) a value to a native bool using BSONElement::trueValue() rules, but with
-         * some types unsupported.  SERVER-6120
-         * @return the bool value
+        /// Add this field to the BSON array under construction.
+        void addToBsonArray(BSONArrayBuilder* pBuilder) const;
+
+        /** Coerce a value to a bool using BSONElement::trueValue() rules.
+         *  Some types unsupported.  SERVER-6120
          */
         bool coerceToBool() const;
 
-        /*
-          Coerce (cast) a value to an int, using JSON rules.
-
-          @returns the int value
-        */
+        /** Coercion operators to extract values with fuzzy type logic.
+         *
+         *  These currently assert if called on an unconvertible type.
+         *  TODO: decided how to handle unsupported types.
+         */
+        string coerceToString() const;
         int coerceToInt() const;
-
-        /*
-          Coerce (cast) a value to a long long, using JSON rules.
-
-          @returns the long value
-        */
         long long coerceToLong() const;
-
-        /*
-          Coerce (cast) a value to a double, using JSON rules.
-
-          @returns the double value
-        */
         double coerceToDouble() const;
-
-        /*
-          Coerce (cast) a value to a date, using JSON rules.
-
-          @returns the date value
-        */
+        OpTime coerceToTimestamp() const;
         long long coerceToDate() const;
         time_t coerceToTimeT() const;
         tm coerceToTm() const; // broken-out time struct (see man gmtime)
 
-        OpTime coerceToTimestamp() const;
-
-        /*
-          Coerce (cast) a value to a string, using JSON rules.
 
-          @returns the date value
-        */
-        string coerceToString() const;
-
-        /*
-          Compare two Values.
-
-          @param rL left value
-          @param rR right value
-          @returns an integer less than zero, zero, or an integer greater than
-            zero, depending on whether rL < rR, rL == rR, or rL > rR
+        /** Compare two Values.
+         *  @returns an integer less than zero, zero, or an integer greater than
+         *           zero, depending on whether lhs < rhs, lhs == rhs, or lhs > rhs
+         *  Warning: may return values other than -1, 0, or 1
+         */
+        static int compare(const Value& lhs, const Value& rhs);
+
+        friend
+        bool operator==(const Value& v1, const Value& v2) {
+            if (v1._storage.identical(v2._storage)) {
+                // Simple case
+                return true;
+            }
+            return (Value::compare(v1, v2) == 0);
+        }
+
+        /// This is for debugging, logging, etc. See getString() for how to extract a string.
+        string toString() const;
+        friend ostream& operator << (ostream& out, const Value& v);
+
+        void swap(Value& rhs) {
+            _storage.swap(rhs._storage);
+        }
+
+        /** Figure out what the widest of two numeric types is.
+         *
+         *  Widest can be thought of as "most capable," or "able to hold the
+         *  largest or most precise value."  The progression is Int, Long, Double.
          */
-        static int compare(const intrusive_ptr<const Value> &rL,
-                           const intrusive_ptr<const Value> &rR);
-
-
-        /*
-          Figure out what the widest of two numeric types is.
-
-          Widest can be thought of as "most capable," or "able to hold the
-          largest or most precise value."  The progression is Int, Long, Double.
-
-          @param rL left value
-          @param rR right value
-          @returns a BSONType of NumberInt, NumberLong, or NumberDouble
-        */
         static BSONType getWidestNumeric(BSONType lType, BSONType rType);
 
-        /*
-          Get the approximate storage size of the value, in bytes.
-
-          @returns approximate storage size of the value.
-         */
+        /// Get the approximate memory size of the value, in bytes. Includes sizeof(Value)
         size_t getApproximateSize() const;
 
-        /*
-          Calculate a hash value.
-
-          Meant to be used to create composite hashes suitable for
-          boost classes such as unordered_map<>.
-
-          @param seed value to augment with this' hash
-        */
-        void hash_combine(size_t &seed) const;
-
-        /*
-          struct Hash is defined to enable the use of Values as
-          keys in boost::unordered_map<>.
+        /** Calculate a hash value.
+         *
+         *  Meant to be used to create composite hashes suitable for
+         *  hashed container classes such as unordered_map<>.
+         */
+        void hash_combine(size_t& seed) const;
 
-          Values are always referenced as immutables in the form
-          intrusive_ptr<const Value>, so these operate on that construction.
-        */
-        struct Hash :
-            unary_function<intrusive_ptr<const Value>, size_t> {
-            size_t operator()(const intrusive_ptr<const Value> &rV) const;
+        /// struct Hash is defined to enable the use of Values as keys in unordered_map.
+        struct Hash : unary_function<const Value&, size_t> {
+            size_t operator()(const Value& rV) const;
         };
 
-    protected:
-        Value(); // creates null value
-        Value(BSONType type); // creates an empty (unitialized value) of type
-                                                // mostly useful for Undefined
-        Value(bool boolValue);
-        Value(int intValue);
-
-    private:
-        Value(BSONElement *pBsonElement);
-
-        Value(long long longValue);
-        Value(double doubleValue);
-        Value(const OpTime& timestampValue);
-        Value(const string &stringValue);
-        Value(const intrusive_ptr<Document> &pDocument);
-        Value(const vector<intrusive_ptr<const Value> > &vpValue);
-
-        void addToBson(Builder *pBuilder) const;
-
-        BSONType type;
-
-        // store values that don't need a ctor/dtor in one of these
-        union {
-            double doubleValue;
-            bool boolValue;
-            int intValue;
-            long long longValue;
-            ReplTime timestampValue;
-            unsigned char oidValue[12];
-            // The member below is redundant, but useful for clarity and searchability.
-            long long dateValue;
-        };
+        /// Call this after memcpying to update ref counts if needed
+        void memcpyed() const { _storage.memcpyed(); }
 
-        string stringValue; // String, Regex, Symbol
-        intrusive_ptr<Document> pDocumentValue;
-        vector<intrusive_ptr<const Value> > vpValue; // for arrays
+        // LEGACY creation functions
+        static Value createFromBsonElement(const BSONElement* pBsonElement);
+        static Value createInt(int value) { return Value(value); }
+        static Value createLong(long long value) { return Value(value); }
+        static Value createDouble(double value) { return Value(value); }
+        static Value createTimestamp(const OpTime& value) { return Value(value); }
+        static Value createString(const string& value) { return Value(value); }
+        static Value createDocument(const Document& doc) { return Value(doc); }
+        static Value createArray(const vector<Value>& vec) { return Value(vec); }
+        static Value createDate(const long long value);
 
-        /*
-        These are often used as the result of boolean or comparison
-        expressions.
+    private:
+        /** This is a "honeypot" to prevent unexpected implicit conversions to the accepted argument
+         *  types. bool is especially bad since without this it will accept any pointer.
+         *
+         *  Template argument name was chosen to make produced error easier to read.
+         */
+        template <typename InvalidArgumentType> Value(const InvalidArgumentType& invalidArgument);
 
-        These are obtained via public static getters defined above.
-        */
-        static const intrusive_ptr<const Value> pFieldUndefined;
-        static const intrusive_ptr<const Value> pFieldNull;
-        static const intrusive_ptr<const Value> pFieldTrue;
-        static const intrusive_ptr<const Value> pFieldFalse;
-        static const intrusive_ptr<const Value> pFieldMinusOne;
-        static const intrusive_ptr<const Value> pFieldZero;
-        static const intrusive_ptr<const Value> pFieldOne;
-
-        /* this implementation is used for getArray() */
-        class vi :
-            public ValueIterator {
-        public:
-            // virtuals from ValueIterator
-            virtual ~vi();
-            virtual bool more() const;
-            virtual intrusive_ptr<const Value> next();
-
-        private:
-            friend class Value;
-            vi(const intrusive_ptr<const Value> &pSource,
-               const vector<intrusive_ptr<const Value> > *pvpValue);
-
-            size_t size;
-            size_t nextIndex;
-            const vector<intrusive_ptr<const Value> > *pvpValue;
-        }; /* class vi */
+        void addToBson(Builder* pBuilder) const;
 
+        ValueStorage _storage;
+        friend class MutableValue; // gets and sets _storage.genericRCPtr
     };
+    BOOST_STATIC_ASSERT(sizeof(Value) == 16);
+}
 
-    /*
-      Equality operator for values.
-
-      Useful for unordered_map<>, etc.
-     */
-    inline bool operator==(const intrusive_ptr<const Value> &v1,
-                    const intrusive_ptr<const Value> &v2) {
-        return (Value::compare(v1, v2) == 0);
-    }
-
-    /*
-      For performance reasons, there are various sharable static values
-      defined in class Value, obtainable by methods such as getUndefined(),
-      getTrue(), getOne(), etc.  We don't want these to go away as they are
-      used by a multitude of threads evaluating pipelines.  In order to avoid
-      having to use atomic integers in the intrusive reference counter, this
-      class overrides the reference counting methods to do nothing, making it
-      safe to use for static Values.
-
-      At this point, only the constructors necessary for the static Values in
-      common use have been defined.  The remainder can be defined if necessary.
-     */
-    class ValueStatic :
-        public Value {
-    public:
-        // virtuals from IntrusiveCounterUnsigned
-        virtual void addRef() const;
-        virtual void release() const;
-
-        // constructors
-        ValueStatic();
-        ValueStatic(BSONType type);
-        ValueStatic(bool boolValue);
-        ValueStatic(int intValue);
-    };
+namespace std {
+    // This is used by std::sort and others
+    template <>
+    inline void swap(mongo::Value& lhs, mongo::Value& rhs) { lhs.swap(rhs); }
 }
 
 /* ======================= INLINED IMPLEMENTATIONS ========================== */
 
 namespace mongo {
 
-    inline BSONType Value::getType() const {
-        return type;
-    }
-
     inline size_t Value::getArrayLength() const {
         verify(getType() == Array);
-        return vpValue.size();
+        return getArray().size();
     }
 
-    inline intrusive_ptr<const Value> Value::getUndefined() {
-        return pFieldUndefined;
+    inline size_t Value::Hash::operator()(const Value& v) const {
+        size_t seed = 0xf0afbeef;
+        v.hash_combine(seed);
+        return seed;
     }
 
-    inline intrusive_ptr<const Value> Value::getNull() {
-        return pFieldNull;
+    inline StringData Value::getStringData() const {
+        verify(getType() == String);
+        return _storage.getString();
     }
 
-    inline intrusive_ptr<const Value> Value::getTrue() {
-        return pFieldTrue;
+    inline string Value::getString() const {
+        verify(getType() == String);
+        StringData sd = _storage.getString();
+        return string(sd.data(), sd.size());
     }
 
-    inline intrusive_ptr<const Value> Value::getFalse() {
-        return pFieldFalse;
+    inline OID Value::getOid() const {
+        verify(getType() == jstOID);
+        return OID(_storage.oid);
     }
 
-    inline intrusive_ptr<const Value> Value::getMinusOne() {
-        return pFieldMinusOne;
+    inline bool Value::getBool() const {
+        verify(getType() == Bool);
+        return _storage.boolValue;
     }
 
-    inline intrusive_ptr<const Value> Value::getZero() {
-        return pFieldZero;
+    inline long long Value::getDate() const {
+        verify(getType() == Date);
+        return _storage.dateValue;
     }
 
-    inline intrusive_ptr<const Value> Value::getOne() {
-        return pFieldOne;
+    inline OpTime Value::getTimestamp() const {
+        verify(getType() == Timestamp);
+        return _storage.timestampValue;
     }
 
-    inline size_t Value::Hash::operator()(
-        const intrusive_ptr<const Value> &rV) const {
-        size_t seed = 0xf0afbeef;
-        rV->hash_combine(seed);
-        return seed;
+    inline string Value::getRegex() const {
+        verify(getType() == RegEx);
+        StringData sd = _storage.getString();
+        return string(sd.data(), sd.size());
     }
 
-    inline ValueStatic::ValueStatic():
-        Value() {
+    inline string Value::getSymbol() const {
+        verify(getType() == Symbol);
+        StringData sd = _storage.getString();
+        return string(sd.data(), sd.size());
     }
 
-    inline ValueStatic::ValueStatic(BSONType type):
-        Value(type) {
+    inline int Value::getInt() const {
+        verify(getType() == NumberInt);
+        return _storage.intValue;
     }
 
-    inline ValueStatic::ValueStatic(bool boolValue):
-        Value(boolValue) {
-    }
+    inline long long Value::getLong() const {
+        BSONType type = getType();
+        if (type == NumberInt)
+            return _storage.intValue;
 
-    inline ValueStatic::ValueStatic(int intValue):
-        Value(intValue) {
+        verify(type == NumberLong);
+        return _storage.longValue;
     }
-
 };
diff --git a/src/mongo/db/pipeline/value_internal.h b/src/mongo/db/pipeline/value_internal.h
new file mode 100644
index 00000000000..578529158a9
--- /dev/null
+++ b/src/mongo/db/pipeline/value_internal.h
@@ -0,0 +1,182 @@
+/**
+ * Copyright (c) 2012 10gen 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/>.
+ */
+
+#pragma once
+
+#include <algorithm>
+#include "bson/bsontypes.h"
+#include "bson/oid.h"
+#include "util/intrusive_counter.h"
+#include "util/optime.h"
+
+
+namespace mongo {
+    class Document;
+    class DocumentStorage;
+    class Value;
+
+    //TODO: a MutableVector, similar to MutableDocument
+    /// A heap-allocated reference-counted std::vector
+    class RCVector : public RefCountable {
+    public:
+        RCVector() {}
+        RCVector(const vector<Value>& v) :vec(v) {}
+        vector<Value> vec;
+    };
+
+#pragma pack(1)
+    class ValueStorage {
+    public:
+        // This is a "missing" Value
+        ValueStorage() { zero(); type = EOO; }
+
+        ValueStorage(BSONType t)                    { zero(); type = t;}
+        ValueStorage(BSONType t, int i)             { zero(); type = t; intValue = i; }
+        ValueStorage(BSONType t, long long l)       { zero(); type = t; longValue = l; }
+        ValueStorage(BSONType t, double d)          { zero(); type = t; doubleValue = d; }
+        ValueStorage(BSONType t, ReplTime r)        { zero(); type = t; timestampValue = r; }
+        ValueStorage(BSONType t, bool b)            { zero(); type = t; boolValue = b; }
+        ValueStorage(BSONType t, const Document& d) { zero(); type = t; putDocument(d); }
+        ValueStorage(BSONType t, const RCVector* a) { zero(); type = t; putVector(a); }
+        ValueStorage(BSONType t, StringData s)      { zero(); type = t; putString(s); }
+
+        ValueStorage(BSONType t, OID& o) {
+            zero();
+            type = t;
+            memcpy(&oid, &o, sizeof(OID));
+            BOOST_STATIC_ASSERT(sizeof(OID) == sizeof(oid));
+        }
+
+        ValueStorage(const ValueStorage& rhs) {
+            memcpy(this, &rhs, sizeof(*this));
+            memcpyed();
+        }
+
+        ~ValueStorage() {
+            if (refCounter)
+                intrusive_ptr_release(genericRCPtr);
+            DEV memset(this, 0xee, sizeof(*this));
+        }
+
+        ValueStorage& operator= (ValueStorage rhsCopy) {
+            this->swap(rhsCopy);
+            return *this;
+        }
+
+        void swap(ValueStorage& rhs) {
+            // Don't need to update ref-counts because they will be the same in the end
+            char temp[sizeof(ValueStorage)];
+            memcpy(temp, this, sizeof(*this));
+            memcpy(this, &rhs, sizeof(*this));
+            memcpy(&rhs, temp, sizeof(*this));
+        }
+
+        /// Call this after memcpying to update ref counts if needed
+        void memcpyed() const {
+            if (refCounter)
+                intrusive_ptr_add_ref(genericRCPtr);
+        }
+
+        /// These are only to be called during Value construction on an empty Value
+        void putString(StringData s);
+        void putVector(const RCVector* v);
+        void putDocument(const Document& d);
+
+        StringData getString() const {
+            if (shortStr) {
+                return StringData(shortStrStorage, shortStrSize);
+            }
+            else {
+                dassert(typeid(*genericRCPtr) == typeid(const RCString));
+                const RCString* stringPtr = static_cast<const RCString*>(genericRCPtr);
+                return StringData(stringPtr->c_str(), stringPtr->size());
+            }
+        }
+
+        const vector<Value>& getArray() const {
+            dassert(typeid(*genericRCPtr) == typeid(const RCVector));
+            const RCVector* arrayPtr = static_cast<const RCVector*>(genericRCPtr);
+            return arrayPtr->vec;
+        }
+
+        // Document is incomplete here so this can't be inline
+        Document getDocument() const;
+
+        BSONType bsonType() const {
+            verify(type != EOO);
+            return type;
+        }
+
+        void zero() {
+            // This is important for identical()
+            memset(this, 0, sizeof(*this));
+        }
+
+        // Byte-for-byte identical
+        bool identical(const ValueStorage& other) const {
+            return  (i64[0] == other.i64[0]
+                  && i64[1] == other.i64[1]);
+        }
+
+        // This data is public because this should only be used by Value which would be a friend
+        union {
+            struct {
+                BSONType type : 8; // one byte (offset 0)
+                union { // one byte (offset 1)
+                    unsigned int flags : 8;
+                    struct {
+                        bool refCounter : 1; // true if we need to refCount
+                        bool shortStr : 1; // true if we are using short strings
+                    };
+                };
+                union { // 14 bytes (offset 2)
+                    unsigned char oid[12];
+                    struct {
+                        int shortStrSize : 8; // TODO Consider moving into flags union (4 bits)
+                        char shortStrStorage[16 - 3]; // ValueStorage is 16 bytes, 3 byte offset
+                    };
+                    struct {
+                        union { // 6 bytes (offset 2)
+                            char pad[6];
+                            char stringCache[6]; // TODO copy first few bytes of strings in here
+                        };
+                        union { // 8 bytes  (offset 8 and aligned)
+                            // There should be no pointers to non-const data
+                            const RefCountable* genericRCPtr;
+
+                            double doubleValue;
+                            bool boolValue;
+                            int intValue;
+                            long long longValue;
+                            ReplTime timestampValue;
+                            long long dateValue;
+                        };
+                    };
+                };
+            };
+            long long i64[2];
+
+            // Note the following are currently unused, but may be useful later
+            int i32[4];
+            unsigned long long u64[2];
+            unsigned int u32[4];
+            char raw[16];
+        };
+    };
+    BOOST_STATIC_ASSERT(sizeof(ValueStorage) == 16);
+#pragma pack()
+
+}