SERVER-17258 Aggregation now supports the reduce expression.

2 files changed, 101 insertions, 0 deletions

diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index 4f1f050478b..9eb2497cd27 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -2572,6 +2572,88 @@ const char* ExpressionRange::getOpName() const {
     return "$range";
 }
 
+/* ------------------------ ExpressionReduce ------------------------------ */
+
+REGISTER_EXPRESSION(reduce, ExpressionReduce::parse);
+intrusive_ptr<Expression> ExpressionReduce::parse(BSONElement expr,
+                                                  const VariablesParseState& vps) {
+    uassert(40075,
+            str::stream() << "$reduce requires an object as an argument, found: "
+                          << typeName(expr.type()),
+            expr.type() == Object);
+
+    intrusive_ptr<ExpressionReduce> reduce(new ExpressionReduce());
+
+    // vpsSub is used only to parse 'in', which must have access to $$this and $$value.
+    VariablesParseState vpsSub(vps);
+    reduce->_thisVar = vpsSub.defineVariable("this");
+    reduce->_valueVar = vpsSub.defineVariable("value");
+
+    for (auto&& elem : expr.Obj()) {
+        auto field = elem.fieldNameStringData();
+
+        if (field == "input") {
+            reduce->_input = parseOperand(elem, vps);
+        } else if (field == "initialValue") {
+            reduce->_initial = parseOperand(elem, vps);
+        } else if (field == "in") {
+            reduce->_in = parseOperand(elem, vpsSub);
+        } else {
+            uasserted(40076, str::stream() << "$reduce found an unknown argument: " << field);
+        }
+    }
+
+    uassert(40077, "$reduce requires 'input' to be specified", reduce->_input);
+    uassert(40078, "$reduce requires 'initialValue' to be specified", reduce->_initial);
+    uassert(40079, "$reduce requires 'in' to be specified", reduce->_in);
+
+    return reduce;
+}
+
+Value ExpressionReduce::evaluateInternal(Variables* vars) const {
+    Value inputVal = _input->evaluateInternal(vars);
+
+    if (inputVal.nullish()) {
+        return Value(BSONNULL);
+    }
+
+    uassert(40080,
+            str::stream() << "$reduce requires that 'input' be an array, found: "
+                          << inputVal.toString(),
+            inputVal.isArray());
+
+    Value accumulatedValue = _initial->evaluateInternal(vars);
+
+    for (auto&& elem : inputVal.getArray()) {
+        vars->setValue(_thisVar, elem);
+        vars->setValue(_valueVar, accumulatedValue);
+
+        accumulatedValue = _in->evaluateInternal(vars);
+    }
+
+    return accumulatedValue;
+}
+
+intrusive_ptr<Expression> ExpressionReduce::optimize() {
+    _input = _input->optimize();
+    _initial = _initial->optimize();
+    _in = _in->optimize();
+    return this;
+}
+
+void ExpressionReduce::addDependencies(DepsTracker* deps, vector<string>* path) const {
+    _input->addDependencies(deps);
+    _initial->addDependencies(deps);
+    _in->addDependencies(deps);
+}
+
+Value ExpressionReduce::serialize(bool explain) const {
+    return Value(Document{{"$reduce",
+                           Document{{"input", _input->serialize(explain)},
+                                    {"initialValue", _initial->serialize(explain)},
+                                    {"in", _in->serialize(explain)}}}});
+}
+
 /* ------------------------ ExpressionReverseArray ------------------------ */
 
 Value ExpressionReverseArray::evaluateInternal(Variables* vars) const {
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index a5c78babce1..1642ee8cb89 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -1135,6 +1135,25 @@ class ExpressionRange final : public ExpressionRangedArity<ExpressionRange, 2, 3
 };
 
 
+class ExpressionReduce final : public Expression {
+public:
+    void addDependencies(DepsTracker* deps, std::vector<std::string>* path = nullptr) const final;
+    Value evaluateInternal(Variables* vars) const final;
+    boost::intrusive_ptr<Expression> optimize() final;
+    static boost::intrusive_ptr<Expression> parse(BSONElement expr,
+                                                  const VariablesParseState& vpsIn);
+    Value serialize(bool explain) const final;
+
+private:
+    boost::intrusive_ptr<Expression> _input;
+    boost::intrusive_ptr<Expression> _initial;
+    boost::intrusive_ptr<Expression> _in;
+
+    Variables::Id _valueVar;
+    Variables::Id _thisVar;
+};
+
+
 class ExpressionSecond final : public ExpressionFixedArity<ExpressionSecond, 1> {
 public:
     Value evaluateInternal(Variables* vars) const final;