SERVER-23029 Added reverseArray expression to aggregation.

5 files changed, 116 insertions, 19 deletions

diff --git a/jstests/aggregation/bugs/reverseArray.js b/jstests/aggregation/bugs/reverseArray.js
new file mode 100644
index 00000000000..8cece4bdcb9
--- /dev/null
+++ b/jstests/aggregation/bugs/reverseArray.js
@@ -0,0 +1,32 @@
+// SERVER-23029 added a new expression, $reverseArray, which consumes an array or a nullish value
+// and produces either the reversed version of that array, or null. In this test file, we check the
+// behavior and error cases.
+load("jstests/aggregation/extras/utils.js"); // For assertErrorCode.
+
+(function() {
+    "use strict";
+
+    var coll = db.reverseArray;
+    coll.drop();
+
+    // We need a document to flow through the pipeline, even though we don't care what fields it
+    // has.
+    coll.insert({});
+
+    assertErrorCode(coll, [{$project: {reversed: {$reverseArray: 1}}}], 34435);
+
+    var res = coll.aggregate([{$project: {reversed: {$reverseArray: {$literal: [1, 2]}}}}]);
+    var output = res.toArray();
+    assert.eq(1, output.length);
+    assert.eq(output[0].reversed, [2, 1]);
+
+    var res = coll.aggregate([{$project: {reversed: {$reverseArray: {$literal: [[1, 2]]}}}}]);
+    var output = res.toArray();
+    assert.eq(1, output.length);
+    assert.eq(output[0].reversed, [[1, 2]]);
+
+    var res = coll.aggregate([{$project: {reversed: {$reverseArray: "$notAField"}}}]);
+    var output = res.toArray();
+    assert.eq(1, output.length);
+    assert.eq(output[0].reversed, null);
+}());
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index b84b529c62e..7cdb9a30fab 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -446,7 +446,7 @@ Value ExpressionAllElementsTrue::evaluateInternal(Variables* vars) const {
     uassert(17040,
             str::stream() << getOpName() << "'s argument must be an array, but is "
                           << typeName(arr.getType()),
-            arr.getType() == Array);
+            arr.isArray());
     const vector<Value>& array = arr.getArray();
     for (vector<Value>::const_iterator it = array.begin(); it != array.end(); ++it) {
         if (!it->coerceToBool()) {
@@ -540,7 +540,7 @@ Value ExpressionAnyElementTrue::evaluateInternal(Variables* vars) const {
     uassert(17041,
             str::stream() << getOpName() << "'s argument must be an array, but is "
                           << typeName(arr.getType()),
-            arr.getType() == Array);
+            arr.isArray());
     const vector<Value>& array = arr.getArray();
     for (vector<Value>::const_iterator it = array.begin(); it != array.end(); ++it) {
         if (it->coerceToBool()) {
@@ -594,7 +594,7 @@ Value ExpressionArrayElemAt::evaluateInternal(Variables* vars) const {
     uassert(28689,
             str::stream() << getOpName() << "'s first argument must be an array, but is "
                           << typeName(array.getType()),
-            array.getType() == Array);
+            array.isArray());
     uassert(28690,
             str::stream() << getOpName() << "'s second argument must be a numeric value,"
                           << " but is " << typeName(indexArg.getType()),
@@ -815,7 +815,7 @@ Value ExpressionConcatArrays::evaluateInternal(Variables* vars) const {
         uassert(28664,
                 str::stream() << "$concatArrays only supports arrays, not "
                               << typeName(val.getType()),
-                val.getType() == Array);
+                val.isArray());
 
         const auto& subValues = val.getArray();
         values.insert(values.end(), subValues.begin(), subValues.end());
@@ -1502,7 +1502,7 @@ void ExpressionFieldPath::addDependencies(DepsTracker* deps, vector<string>* pat
 }
 
 Value ExpressionFieldPath::evaluatePathArray(size_t index, const Value& input) const {
-    dassert(input.getType() == Array);
+    dassert(input.isArray());
 
     // Check for remaining path in each element of array
     vector<Value> result;
@@ -1644,7 +1644,7 @@ Value ExpressionFilter::evaluateInternal(Variables* vars) const {
     uassert(28651,
             str::stream() << "input to $filter must be an array not "
                           << typeName(inputVal.getType()),
-            inputVal.getType() == Array);
+            inputVal.isArray());
 
     const vector<Value>& input = inputVal.getArray();
 
@@ -1847,7 +1847,7 @@ Value ExpressionMap::evaluateInternal(Variables* vars) const {
 
     uassert(16883,
             str::stream() << "input to $map must be an array not " << typeName(inputVal.getType()),
-            inputVal.getType() == Array);
+            inputVal.isArray());
 
     const vector<Value>& input = inputVal.getArray();
 
@@ -2499,6 +2499,34 @@ const char* ExpressionPow::getOpName() const {
     return "$pow";
 }
 
+/* ------------------------ ExpressionReverseArray ------------------------ */
+
+Value ExpressionReverseArray::evaluateInternal(Variables* vars) const {
+    Value input(vpOperand[0]->evaluateInternal(vars));
+
+    if (input.nullish()) {
+        return Value(BSONNULL);
+    }
+
+    uassert(34435,
+            str::stream() << "The argument to $reverseArray must be an array, but was of type: "
+                          << typeName(input.getType()),
+            input.isArray());
+
+    if (input.getArrayLength() < 2) {
+        return input;
+    }
+
+    std::vector<Value> array = input.getArray();
+    std::reverse(array.begin(), array.end());
+    return Value(array);
+}
+
+REGISTER_EXPRESSION(reverseArray, ExpressionReverseArray::parse);
+const char* ExpressionReverseArray::getOpName() const {
+    return "$reverseArray";
+}
+
 /* ------------------------- ExpressionSecond ----------------------------- */
 
 Value ExpressionSecond::evaluateInternal(Variables* vars) const {
@@ -2531,11 +2559,11 @@ Value ExpressionSetDifference::evaluateInternal(Variables* vars) const {
     uassert(17048,
             str::stream() << "both operands of $setDifference must be arrays. First "
                           << "argument is of type: " << typeName(lhs.getType()),
-            lhs.getType() == Array);
+            lhs.isArray());
     uassert(17049,
             str::stream() << "both operands of $setDifference must be arrays. Second "
                           << "argument is of type: " << typeName(rhs.getType()),
-            rhs.getType() == Array);
+            rhs.isArray());
 
     ValueSet rhsSet = arrayToSet(rhs);
     const vector<Value>& lhsArray = lhs.getArray();
@@ -2573,7 +2601,7 @@ Value ExpressionSetEquals::evaluateInternal(Variables* vars) const {
         uassert(17044,
                 str::stream() << "All operands of $setEquals must be arrays. One "
                               << "argument is of type: " << typeName(nextEntry.getType()),
-                nextEntry.getType() == Array);
+                nextEntry.isArray());
 
         if (i == 0) {
             lhs.insert(nextEntry.getArray().begin(), nextEntry.getArray().end());
@@ -2605,7 +2633,7 @@ Value ExpressionSetIntersection::evaluateInternal(Variables* vars) const {
         uassert(17047,
                 str::stream() << "All operands of $setIntersection must be arrays. One "
                               << "argument is of type: " << typeName(nextEntry.getType()),
-                nextEntry.getType() == Array);
+                nextEntry.isArray());
 
         if (i == 0) {
             currentIntersection.insert(nextEntry.getArray().begin(), nextEntry.getArray().end());
@@ -2660,11 +2688,11 @@ Value ExpressionSetIsSubset::evaluateInternal(Variables* vars) const {
     uassert(17046,
             str::stream() << "both operands of $setIsSubset must be arrays. First "
                           << "argument is of type: " << typeName(lhs.getType()),
-            lhs.getType() == Array);
+            lhs.isArray());
     uassert(17042,
             str::stream() << "both operands of $setIsSubset must be arrays. Second "
                           << "argument is of type: " << typeName(rhs.getType()),
-            rhs.getType() == Array);
+            rhs.isArray());
 
     return setIsSubsetHelper(lhs.getArray(), arrayToSet(rhs));
 }
@@ -2689,7 +2717,7 @@ public:
         uassert(17310,
                 str::stream() << "both operands of $setIsSubset must be arrays. First "
                               << "argument is of type: " << typeName(lhs.getType()),
-                lhs.getType() == Array);
+                lhs.isArray());
 
         return setIsSubsetHelper(lhs.getArray(), _cachedRhsSet);
     }
@@ -2711,7 +2739,7 @@ intrusive_ptr<Expression> ExpressionSetIsSubset::optimize() {
         uassert(17311,
                 str::stream() << "both operands of $setIsSubset must be arrays. Second "
                               << "argument is of type: " << typeName(rhs.getType()),
-                rhs.getType() == Array);
+                rhs.isArray());
 
         return new Optimized(arrayToSet(rhs), vpOperand);
     }
@@ -2736,7 +2764,7 @@ Value ExpressionSetUnion::evaluateInternal(Variables* vars) const {
         uassert(17043,
                 str::stream() << "All operands of $setUnion must be arrays. One argument"
                               << " is of type: " << typeName(newEntries.getType()),
-                newEntries.getType() == Array);
+                newEntries.isArray());
 
         unionedSet.insert(newEntries.getArray().begin(), newEntries.getArray().end());
     }
@@ -2752,7 +2780,7 @@ const char* ExpressionSetUnion::getOpName() const {
 
 Value ExpressionIsArray::evaluateInternal(Variables* vars) const {
     Value argument = vpOperand[0]->evaluateInternal(vars);
-    return Value(argument.getType() == Array);
+    return Value(argument.isArray());
 }
 
 REGISTER_EXPRESSION(isArray, ExpressionIsArray::parse);
@@ -2776,7 +2804,7 @@ Value ExpressionSlice::evaluateInternal(Variables* vars) const {
     uassert(28724,
             str::stream() << "First argument to $slice must be an array, but is"
                           << " of type: " << typeName(arrayVal.getType()),
-            arrayVal.getType() == Array);
+            arrayVal.isArray());
     uassert(28725,
             str::stream() << "Second argument to $slice must be a numeric value,"
                           << " but is of type: " << typeName(arg2.getType()),
@@ -2854,7 +2882,7 @@ Value ExpressionSize::evaluateInternal(Variables* vars) const {
     uassert(17124,
             str::stream() << "The argument to $size must be an array, but was of type: "
                           << typeName(array.getType()),
-            array.getType() == Array);
+            array.isArray());
     return Value::createIntOrLong(array.getArray().size());
 }
 
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index d0f1370daf7..07ba3f23ff2 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -1201,6 +1201,13 @@ public:
 };
 
 
+class ExpressionReverseArray final : public ExpressionFixedArity<ExpressionReverseArray, 1> {
+public:
+    Value evaluateInternal(Variables* vars) const final;
+    const char* getOpName() const final;
+};
+
+
 class ExpressionSlice final : public ExpressionRangedArity<ExpressionSlice, 2, 3> {
 public:
     Value evaluateInternal(Variables* vars) const final;
diff --git a/src/mongo/db/pipeline/expression_test.cpp b/src/mongo/db/pipeline/expression_test.cpp
index 63a6375e85a..a9ccb39c58b 100644
--- a/src/mongo/db/pipeline/expression_test.cpp
+++ b/src/mongo/db/pipeline/expression_test.cpp
@@ -29,6 +29,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/bson/bsonmisc.h"
+#include "mongo/config.h"
 #include "mongo/db/pipeline/accumulator.h"
 #include "mongo/db/pipeline/document.h"
 #include "mongo/db/pipeline/expression.h"
@@ -756,6 +757,28 @@ TEST_F(ExpressionFloorTest, NullArg) {
     assertEvaluates(Value(BSONNULL), Value(BSONNULL));
 }
 
+/* ------------------------ ExpressionReverseArray -------------------- */
+
+TEST(ExpressionReverseArrayTest, ReversesNormalArray) {
+    assertExpectedResults("$reverseArray",
+                          {{{Value(BSON_ARRAY(1 << 2 << 3))}, Value(BSON_ARRAY(3 << 2 << 1))}});
+}
+
+TEST(ExpressionReverseArrayTest, ReversesEmptyArray) {
+    assertExpectedResults("$reverseArray",
+                          {{{Value(std::vector<Value>())}, Value(std::vector<Value>())}});
+}
+
+TEST(ExpressionReverseArrayTest, ReversesOneElementArray) {
+    assertExpectedResults("$reverseArray", {{{Value(BSON_ARRAY(1))}, Value(BSON_ARRAY(1))}});
+}
+
+TEST(ExpressionReverseArrayTest, ReturnsNullWithNullishInput) {
+    assertExpectedResults(
+        "$reverseArray",
+        {{{Value(BSONNULL)}, Value(BSONNULL)}, {{Value(BSONUndefined)}, Value(BSONNULL)}});
+}
+
 /* ------------------------- ExpressionTrunc -------------------------- */
 
 class ExpressionTruncTest : public ExpressionNaryTestOneArg {
diff --git a/src/mongo/db/pipeline/value.h b/src/mongo/db/pipeline/value.h
index 036c6c41876..9fc51e164b2 100644
--- a/src/mongo/db/pipeline/value.h
+++ b/src/mongo/db/pipeline/value.h
@@ -137,6 +137,13 @@ public:
     }
 
     /**
+     * Return true if the Value is an array.
+     */
+    bool isArray() const {
+        return _storage.type == Array;
+    }
+
+    /**
      * Returns true if this value is a numeric type that can be represented as a 32-bit integer,
      * and false otherwise.
      */