Accumulators are not Expressions so they shouldn't derive

There may be a place for a common parent of "tree-like" things, but
Expression is not the correct base. This commit keeps the public API of
accumulators the same so consumers don't need to be modified.

11 files changed, 109 insertions, 210 deletions

diff --git a/src/mongo/SConscript b/src/mongo/SConscript
index 14f45278093..86298ca6408 100644
--- a/src/mongo/SConscript
+++ b/src/mongo/SConscript
@@ -307,7 +307,6 @@ env.StaticLibrary("coredb", [
         "db/pipeline/accumulator_last.cpp",
         "db/pipeline/accumulator_min_max.cpp",
         "db/pipeline/accumulator_push.cpp",
-        "db/pipeline/accumulator_single_value.cpp",
         "db/pipeline/accumulator_sum.cpp",
         "db/pipeline/builder.cpp",
         "db/pipeline/doc_mem_monitor.cpp",
diff --git a/src/mongo/db/pipeline/accumulator.cpp b/src/mongo/db/pipeline/accumulator.cpp
index 13f21da064c..db6024664a4 100644
--- a/src/mongo/db/pipeline/accumulator.cpp
+++ b/src/mongo/db/pipeline/accumulator.cpp
@@ -25,25 +25,8 @@
 namespace mongo {
     using namespace mongoutils;
 
-    void Accumulator::addOperand(
-        const intrusive_ptr<Expression> &pExpression) {
-        uassert(15943, str::stream() << "group accumulator " <<
-                getOpName() << " only accepts one operand",
-                vpOperand.size() < 1);
-        
-        ExpressionNary::addOperand(pExpression);
-    }
-
-    Accumulator::Accumulator():
-        ExpressionNary() {
-    }
-
     Value Accumulator::serialize() const {
-        verify(vpOperand.size() == 1);
-
-        MutableDocument valBuilder;
-        valBuilder[getOpName()] = vpOperand[0]->serialize();
-        return valBuilder.freezeToValue();
+        return Value(DOC(getOpName() << _expr->serialize()));
     }
 
     void agg_framework_reservedErrors() {
diff --git a/src/mongo/db/pipeline/accumulator.h b/src/mongo/db/pipeline/accumulator.h
index 3c1c12464c7..dcbf7d33820 100644
--- a/src/mongo/db/pipeline/accumulator.h
+++ b/src/mongo/db/pipeline/accumulator.h
@@ -26,44 +26,47 @@
 namespace mongo {
     class ExpressionContext;
 
-    class Accumulator :
-        public ExpressionNary {
+    class Accumulator : public RefCountable {
     public:
-        // virtuals from ExpressionNary
-        virtual void addOperand(const intrusive_ptr<Expression> &pExpression);
-        virtual Value serialize() const;
+        /// Serialize Accumulator and its input generating Expression.
+        Value serialize() const;
 
-        /*
-          Get the accumulated value.
+        // TODO rename to process()
+        /// Process input and update internal state.
+        void evaluate(const Document& input) { processInternal(_expr->evaluate(input)); }
 
-          @returns the accumulated value
-         */
+        /// Add needed fields to the passed in set
+        void addDependencies(set<string>& deps) const { _expr->addDependencies(deps); }
+
+        // TODO move into constructor
+        /// Sets the expression used to generate input to the accumulator.
+        void addOperand(const intrusive_ptr<Expression>& expr) {
+            verify(!_expr); // only takes one operand
+            _expr = expr;
+        }
+
+        /// Marks the end of the evaluate() phase and return accumulated result.
         virtual Value getValue() const = 0;
 
+        /// The name of the op as used in a serialization of the pipeline.
+        virtual const char* getOpName() const = 0;
+
     protected:
-        Accumulator();
+        Accumulator() {}
 
-        /*
-          Convenience method for doing this for accumulators.  The pattern
-          is always the same, so a common implementation works, but requires
-          knowing the operator name.
+        /// Update subclass's internal state based on input
+        virtual void processInternal(const Value& input) = 0;
 
-          @param pBuilder the builder to add to
-          @param fieldName the projected name
-          @param opName the operator name
-         */
-        void opToBson(BSONObjBuilder *pBuilder, StringData opName,
-                      StringData fieldName, bool requireExpression) const;
+    private:
+        intrusive_ptr<Expression> _expr;
     };
 
 
-    class AccumulatorAddToSet :
-        public Accumulator {
+    class AccumulatorAddToSet : public Accumulator {
     public:
-        // virtuals from Expression
-        virtual Value evaluate(const Document& pDocument) const;
+        virtual void processInternal(const Value& input);
         virtual Value getValue() const;
-        virtual const char *getOpName() const;
+        virtual const char* getOpName() const;
 
         /*
           Create an appending accumulator.
@@ -76,38 +79,17 @@ namespace mongo {
 
     private:
         AccumulatorAddToSet(const intrusive_ptr<ExpressionContext> &pTheCtx);
-        typedef boost::unordered_set<Value, Value::Hash > SetType;
-        mutable SetType set;
-        mutable SetType::iterator itr; 
+        typedef boost::unordered_set<Value, Value::Hash> SetType;
+        SetType set;
         intrusive_ptr<ExpressionContext> pCtx;
     };
 
 
-    /*
-      This isn't a finished accumulator, but rather a convenient base class
-      for others such as $first, $last, $max, $min, and similar.  It just
-      provides a holder for a single Value, and the getter for that.  The
-      holder is protected so derived classes can manipulate it.
-     */
-    class AccumulatorSingleValue :
-        public Accumulator {
+    class AccumulatorFirst : public Accumulator {
     public:
-        // virtuals from Expression
+        virtual void processInternal(const Value& input);
         virtual Value getValue() const;
-
-    protected:
-        AccumulatorSingleValue();
-
-        mutable Value pValue; /* current min/max */
-    };
-
-
-    class AccumulatorFirst :
-        public AccumulatorSingleValue {
-    public:
-        // virtuals from Expression
-        virtual Value evaluate(const Document& pDocument) const;
-        virtual const char *getOpName() const;
+        virtual const char* getOpName() const;
 
         /*
           Create the accumulator.
@@ -118,17 +100,18 @@ namespace mongo {
             const intrusive_ptr<ExpressionContext> &pCtx);
 
     private:
-        mutable bool _haveFirst;
         AccumulatorFirst();
+
+        bool _haveFirst;
+        Value _first;
     };
 
 
-    class AccumulatorLast :
-        public AccumulatorSingleValue {
+    class AccumulatorLast : public Accumulator {
     public:
-        // virtuals from Expression
-        virtual Value evaluate(const Document& pDocument) const;
-        virtual const char *getOpName() const;
+        virtual void processInternal(const Value& input);
+        virtual Value getValue() const;
+        virtual const char* getOpName() const;
 
         /*
           Create the accumulator.
@@ -140,16 +123,15 @@ namespace mongo {
 
     private:
         AccumulatorLast();
+        Value _last;
     };
 
 
-    class AccumulatorSum :
-        public Accumulator {
+    class AccumulatorSum : public Accumulator {
     public:
-        // virtuals from Accumulator
-        virtual Value evaluate(const Document& pDocument) const;
+        virtual void processInternal(const Value& input);
         virtual Value getValue() const;
-        virtual const char *getOpName() const;
+        virtual const char* getOpName() const;
 
         /*
           Create a summing accumulator.
@@ -163,20 +145,19 @@ namespace mongo {
     protected: /* reused by AccumulatorAvg */
         AccumulatorSum();
 
-        mutable BSONType totalType;
-        mutable long long longTotal;
-        mutable double doubleTotal;
+        BSONType totalType;
+        long long longTotal;
+        double doubleTotal;
         // count is only used by AccumulatorAvg, but lives here to avoid counting non-numeric values
-        mutable long long count;
+        long long count;
     };
 
 
-    class AccumulatorMinMax :
-        public AccumulatorSingleValue {
+    class AccumulatorMinMax : public Accumulator {
     public:
-        // virtuals from Expression
-        virtual Value evaluate(const Document& pDocument) const;
-        virtual const char *getOpName() const;
+        virtual void processInternal(const Value& input);
+        virtual Value getValue() const;
+        virtual const char* getOpName() const;
 
         /*
           Create either the max or min accumulator.
@@ -191,17 +172,16 @@ namespace mongo {
     private:
         AccumulatorMinMax(int theSense);
 
-        int sense; /* 1 for min, -1 for max; used to "scale" comparison */
+        Value _val;
+        const int _sense; /* 1 for min, -1 for max; used to "scale" comparison */
     };
 
 
-    class AccumulatorPush :
-        public Accumulator {
+    class AccumulatorPush : public Accumulator {
     public:
-        // virtuals from Expression
-        virtual Value evaluate(const Document& pDocument) const;
+        virtual void processInternal(const Value& input);
         virtual Value getValue() const;
-        virtual const char *getOpName() const;
+        virtual const char* getOpName() const;
 
         /*
           Create an appending accumulator.
@@ -215,19 +195,17 @@ namespace mongo {
     private:
         AccumulatorPush(const intrusive_ptr<ExpressionContext> &pTheCtx);
 
-        mutable vector<Value> vpValue;
+        vector<Value> vpValue;
         intrusive_ptr<ExpressionContext> pCtx;
     };
 
 
-    class AccumulatorAvg :
-        public AccumulatorSum {
+    class AccumulatorAvg : public AccumulatorSum {
         typedef AccumulatorSum Super;
     public:
-        // virtuals from Accumulator
-        virtual Value evaluate(const Document& pDocument) const;
+        virtual void processInternal(const Value& input);
         virtual Value getValue() const;
-        virtual const char *getOpName() const;
+        virtual const char* getOpName() const;
 
         /*
           Create an averaging accumulator.
diff --git a/src/mongo/db/pipeline/accumulator_add_to_set.cpp b/src/mongo/db/pipeline/accumulator_add_to_set.cpp
index 7af52df062b..aaf2fc4e6a5 100644
--- a/src/mongo/db/pipeline/accumulator_add_to_set.cpp
+++ b/src/mongo/db/pipeline/accumulator_add_to_set.cpp
@@ -21,13 +21,10 @@
 #include "db/pipeline/value.h"
 
 namespace mongo {
-    Value AccumulatorAddToSet::evaluate(const Document& pDocument) const {
-        verify(vpOperand.size() == 1);
-        Value prhs(vpOperand[0]->evaluate(pDocument));
-
+    void AccumulatorAddToSet::processInternal(const Value& input) {
         if (!pCtx->getDoingMerge()) {
-            if (!prhs.missing()) {
-                set.insert(prhs);
+            if (!input.missing()) {
+                set.insert(input);
             }
         } else {
             /*
@@ -36,23 +33,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(input.getType() == Array);
             
-            const vector<Value>& array = prhs.getArray();
+            const vector<Value>& array = input.getArray();
             set.insert(array.begin(), array.end());
         }
-
-        return Value();
     }
 
     Value AccumulatorAddToSet::getValue() const {
-        vector<Value> valVec;
-
-        for (itr = set.begin(); itr != set.end(); ++itr) {
-            valVec.push_back(*itr);
-        }
-
-        return Value(valVec);
+        vector<Value> valVec(set.begin(), set.end());
+        return Value::consume(valVec);
     }
 
     AccumulatorAddToSet::AccumulatorAddToSet(
diff --git a/src/mongo/db/pipeline/accumulator_avg.cpp b/src/mongo/db/pipeline/accumulator_avg.cpp
index a626ec93e87..abfd8ede788 100644
--- a/src/mongo/db/pipeline/accumulator_avg.cpp
+++ b/src/mongo/db/pipeline/accumulator_avg.cpp
@@ -26,9 +26,9 @@ namespace mongo {
     const char AccumulatorAvg::subTotalName[] = "subTotal";
     const char AccumulatorAvg::countName[] = "count";
 
-    Value AccumulatorAvg::evaluate(const Document& pDocument) const {
+    void AccumulatorAvg::processInternal(const Value& input) {
         if (!pCtx->getDoingMerge()) {
-            Super::evaluate(pDocument);
+            Super::processInternal(input);
         }
         else {
             /*
@@ -36,19 +36,16 @@ namespace mongo {
               both a subtotal and a count.  This is what getValue() produced
               below.
              */
-            Value shardOut = vpOperand[0]->evaluate(pDocument);
-            verify(shardOut.getType() == Object);
+            verify(input.getType() == Object);
 
-            Value subTotal = shardOut[subTotalName];
+            Value subTotal = input[subTotalName];
             verify(!subTotal.missing());
             doubleTotal += subTotal.getDouble();
                 
-            Value subCount = shardOut[countName];
+            Value subCount = input[countName];
             verify(!subCount.missing());
             count += subCount.getLong();
         }
-
-        return Value();
     }
 
     intrusive_ptr<Accumulator> AccumulatorAvg::create(
diff --git a/src/mongo/db/pipeline/accumulator_first.cpp b/src/mongo/db/pipeline/accumulator_first.cpp
index ce53cccd365..3b26e5b5705 100644
--- a/src/mongo/db/pipeline/accumulator_first.cpp
+++ b/src/mongo/db/pipeline/accumulator_first.cpp
@@ -21,22 +21,21 @@
 
 namespace mongo {
 
-    Value AccumulatorFirst::evaluate(const Document& pDocument) const {
-        verify(vpOperand.size() == 1);
-
+    void AccumulatorFirst::processInternal(const Value& input) {
         /* only remember the first value seen */
         if (!_haveFirst) {
             // can't use pValue.missing() since we want the first value even if missing
             _haveFirst = true;
-            pValue = vpOperand[0]->evaluate(pDocument);
+            _first = input;
         }
+    }
 
-        return pValue;
+    Value AccumulatorFirst::getValue() const {
+        return _first;
     }
 
     AccumulatorFirst::AccumulatorFirst()
-        : AccumulatorSingleValue()
-        , _haveFirst(false)
+        : _haveFirst(false)
     {}
 
     intrusive_ptr<Accumulator> AccumulatorFirst::create(
diff --git a/src/mongo/db/pipeline/accumulator_last.cpp b/src/mongo/db/pipeline/accumulator_last.cpp
index ea65094d04b..c6fbe859719 100644
--- a/src/mongo/db/pipeline/accumulator_last.cpp
+++ b/src/mongo/db/pipeline/accumulator_last.cpp
@@ -21,19 +21,17 @@
 
 namespace mongo {
 
-    Value AccumulatorLast::evaluate(const Document& pDocument) const {
-        verify(vpOperand.size() == 1);
-
+    void AccumulatorLast::processInternal(const Value& input) {
         /* always remember the last value seen */
-        pValue = vpOperand[0]->evaluate(pDocument);
-
-        return pValue;
+        _last = input;
     }
 
-    AccumulatorLast::AccumulatorLast():
-        AccumulatorSingleValue() {
+    Value AccumulatorLast::getValue() const {
+        return _last;
     }
 
+    AccumulatorLast::AccumulatorLast() {}
+
     intrusive_ptr<Accumulator> AccumulatorLast::create(
         const intrusive_ptr<ExpressionContext> &pCtx) {
         intrusive_ptr<AccumulatorLast> pAccumulator(
diff --git a/src/mongo/db/pipeline/accumulator_min_max.cpp b/src/mongo/db/pipeline/accumulator_min_max.cpp
index 86172cb3338..c3c5a9790f0 100644
--- a/src/mongo/db/pipeline/accumulator_min_max.cpp
+++ b/src/mongo/db/pipeline/accumulator_min_max.cpp
@@ -21,27 +21,24 @@
 
 namespace mongo {
 
-    Value AccumulatorMinMax::evaluate(const Document& pDocument) const {
-        verify(vpOperand.size() == 1);
-        Value prhs(vpOperand[0]->evaluate(pDocument));
-
+    void AccumulatorMinMax::processInternal(const Value& input) {
         // nullish values should have no impact on result
-        if (!prhs.nullish()) {
+        if (!input.nullish()) {
             /* compare with the current value; swap if appropriate */
-            int cmp = Value::compare(pValue, prhs) * sense;
-            if (cmp > 0 || pValue.missing()) // missing is lower than all other values
-                pValue = prhs;
+            int cmp = Value::compare(_val, input) * _sense;
+            if (cmp > 0 || _val.missing()) // missing is lower than all other values
+                _val = input;
         }
-
-        return Value();
     }
 
-    AccumulatorMinMax::AccumulatorMinMax(int theSense):
-        AccumulatorSingleValue(),
-        sense(theSense) {
-        verify((sense == 1) || (sense == -1));
+    Value AccumulatorMinMax::getValue() const {
+        return _val;
     }
 
+    AccumulatorMinMax::AccumulatorMinMax(int theSense)
+        :_sense(theSense)
+    { verify((_sense == 1) || (_sense == -1)); }
+
     intrusive_ptr<Accumulator> AccumulatorMinMax::createMin(
         const intrusive_ptr<ExpressionContext> &pCtx) {
         intrusive_ptr<AccumulatorMinMax> pAccumulator(
@@ -57,7 +54,7 @@ namespace mongo {
     }
 
     const char *AccumulatorMinMax::getOpName() const {
-        if (sense == 1)
+        if (_sense == 1)
             return "$min";
         return "$max";
     }
diff --git a/src/mongo/db/pipeline/accumulator_push.cpp b/src/mongo/db/pipeline/accumulator_push.cpp
index 640f627708b..55f914dd888 100644
--- a/src/mongo/db/pipeline/accumulator_push.cpp
+++ b/src/mongo/db/pipeline/accumulator_push.cpp
@@ -21,13 +21,10 @@
 #include "db/pipeline/value.h"
 
 namespace mongo {
-    Value AccumulatorPush::evaluate(const Document& pDocument) const {
-        verify(vpOperand.size() == 1);
-        Value prhs(vpOperand[0]->evaluate(pDocument));
-
+    void AccumulatorPush::processInternal(const Value& input) {
         if (!pCtx->getDoingMerge()) {
-            if (!prhs.missing()) {
-                vpValue.push_back(prhs);
+            if (!input.missing()) {
+                vpValue.push_back(input);
             }
         }
         else {
@@ -37,13 +34,11 @@ 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(input.getType() == Array);
             
-            const vector<Value>& vec = prhs.getArray();
+            const vector<Value>& vec = input.getArray();
             vpValue.insert(vpValue.end(), vec.begin(), vec.end());
         }
-
-        return Value();
     }
 
     Value AccumulatorPush::getValue() const {
diff --git a/src/mongo/db/pipeline/accumulator_single_value.cpp b/src/mongo/db/pipeline/accumulator_single_value.cpp
deleted file mode 100644
index 76187d52222..00000000000
--- a/src/mongo/db/pipeline/accumulator_single_value.cpp
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * Copyright (c) 2011 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/>.
- */
-
-#include "pch.h"
-#include "accumulator.h"
-
-#include "db/pipeline/value.h"
-
-namespace mongo {
-
-    Value AccumulatorSingleValue::getValue() const {
-        return pValue;
-    }
-
-    AccumulatorSingleValue::AccumulatorSingleValue():
-        pValue(Value()) {
-    }
-
-}
diff --git a/src/mongo/db/pipeline/accumulator_sum.cpp b/src/mongo/db/pipeline/accumulator_sum.cpp
index bcf8195efd4..9bc6b9fc507 100644
--- a/src/mongo/db/pipeline/accumulator_sum.cpp
+++ b/src/mongo/db/pipeline/accumulator_sum.cpp
@@ -21,24 +21,21 @@
 
 namespace mongo {
 
-    Value AccumulatorSum::evaluate(const Document& pDocument) const {
-        verify(vpOperand.size() == 1);
-        Value rhs = vpOperand[0]->evaluate(pDocument);
-
+    void AccumulatorSum::processInternal(const Value& input) {
         // do nothing with non numeric types
-        if (!rhs.numeric())
-            return Value();
+        if (!input.numeric())
+            return;
 
         // upgrade to the widest type required to hold the result
-        totalType = Value::getWidestNumeric(totalType, rhs.getType());
+        totalType = Value::getWidestNumeric(totalType, input.getType());
 
         if (totalType == NumberInt || totalType == NumberLong) {
-            long long v = rhs.coerceToLong();
+            long long v = input.coerceToLong();
             longTotal += v;
             doubleTotal += v;
         }
         else if (totalType == NumberDouble) {
-            double v = rhs.coerceToDouble();
+            double v = input.coerceToDouble();
             doubleTotal += v;
         }
         else {
@@ -47,8 +44,6 @@ namespace mongo {
         }
 
         count++;
-
-        return Value();
     }
 
     intrusive_ptr<Accumulator> AccumulatorSum::create(