SERVER-57553 add $sum accumulator to SBE

7 files changed, 631 insertions, 1 deletions

diff --git a/src/mongo/db/exec/sbe/expressions/expression.cpp b/src/mongo/db/exec/sbe/expressions/expression.cpp
index c85f035b48f..52b8a6a3b42 100644
--- a/src/mongo/db/exec/sbe/expressions/expression.cpp
+++ b/src/mongo/db/exec/sbe/expressions/expression.cpp
@@ -521,6 +521,11 @@ static stdx::unordered_map<std::string, InstrFn> kInstrFunctions = {
     {"isTimestamp",
      InstrFn{[](size_t n) { return n == 1; }, &vm::CodeFragment::appendIsTimestamp, false}},
     {"sum", InstrFn{[](size_t n) { return n == 1; }, &vm::CodeFragment::appendSum, true}},
+    {"aggDoubleDoubleSum",
+     InstrFn{[](size_t n) { return n == 1; }, &vm::CodeFragment::appendDoubleDoubleSum, true}},
+    {"doubleDoubleSumFinalize",
+     InstrFn{
+         [](size_t n) { return n == 1; }, &vm::CodeFragment::appendDoubleDoubleSumFinalize, false}},
     {"min", InstrFn{[](size_t n) { return n == 1; }, &vm::CodeFragment::appendMin, true}},
     {"max", InstrFn{[](size_t n) { return n == 1; }, &vm::CodeFragment::appendMax, true}},
     {"first", InstrFn{[](size_t n) { return n == 1; }, &vm::CodeFragment::appendFirst, true}},
diff --git a/src/mongo/db/exec/sbe/vm/arith.cpp b/src/mongo/db/exec/sbe/vm/arith.cpp
index a8528fc7877..ce35e486619 100644
--- a/src/mongo/db/exec/sbe/vm/arith.cpp
+++ b/src/mongo/db/exec/sbe/vm/arith.cpp
@@ -32,6 +32,7 @@
 #include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/platform/overflow_arithmetic.h"
 #include "mongo/util/represent_as.h"
+#include "mongo/util/summation.h"
 #include "mongo/util/time_support.h"
 
 namespace mongo {
@@ -376,6 +377,142 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::genericAdd(value::Type
     return genericArithmeticOp<Addition>(lhsTag, lhsValue, rhsTag, rhsValue);
 }
 
+namespace {
+Array* reserveArrayCapacity(Value arrVal, size_t cap) {
+    auto arr = reinterpret_cast<Array*>(arrVal);
+    arr->reserve(cap);
+
+    return arr;
+}
+
+void setNonDecimalTotal(TypeTags nonDecimalTotalTag,
+                        const DoubleDoubleSummation& nonDecimalTotal,
+                        Array* arr) {
+    auto [sum, addend] = nonDecimalTotal.getDoubleDouble();
+    arr->push_back(nonDecimalTotalTag, value::bitcastFrom<int32_t>(0.0));
+    arr->push_back(TypeTags::NumberDouble, value::bitcastFrom<double>(sum));
+    arr->push_back(TypeTags::NumberDouble, value::bitcastFrom<double>(addend));
+}
+
+std::tuple<bool, value::TypeTags, value::Value> makeSumResultArray(
+    TypeTags nonDecimalTotalTag, const DoubleDoubleSummation& nonDecimalTotal) {
+    auto [retTag, retVal] = makeNewArray();
+    ValueGuard guard{retTag, retVal};
+    setNonDecimalTotal(nonDecimalTotalTag,
+                       nonDecimalTotal,
+                       reserveArrayCapacity(retVal, AggSumValueElems::kMaxSizeOfArray - 1));
+
+    guard.reset();
+    return {true, TypeTags::Array, retVal};
+}
+
+std::tuple<bool, value::TypeTags, value::Value> makeSumResultArray(
+    TypeTags nonDecimalTotalTag,
+    const DoubleDoubleSummation& nonDecimalTotal,
+    const Decimal128& decimalTotal) {
+    auto [retTag, retVal] = makeNewArray();
+    ValueGuard guard{retTag, retVal};
+    auto ret = reserveArrayCapacity(retVal, AggSumValueElems::kMaxSizeOfArray);
+    setNonDecimalTotal(nonDecimalTotalTag, nonDecimalTotal, ret);
+    // We don't need to use 'ValueGuard' for decimal because we've already allocated enough storage
+    // with `reserveArrayCapacity` and Array::push_back() is guaranteed to not throw.
+    auto [tag, val] = makeCopyDecimal(decimalTotal);
+    ret->push_back(tag, val);
+
+    guard.reset();
+    return {true, TypeTags::Array, retVal};
+}
+
+void addNonDecimal(TypeTags tag, Value val, DoubleDoubleSummation& nonDecimalTotal) {
+    switch (tag) {
+        case TypeTags::NumberInt64:
+            nonDecimalTotal.addLong(value::bitcastTo<int64_t>(val));
+            break;
+        case TypeTags::NumberInt32:
+            nonDecimalTotal.addInt(value::bitcastTo<int32_t>(val));
+            break;
+        case TypeTags::NumberDouble:
+            nonDecimalTotal.addDouble(value::bitcastTo<double>(val));
+            break;
+        default:
+            MONGO_UNREACHABLE_TASSERT(5755316);
+    }
+}
+}  // namespace
+
+std::tuple<bool, value::TypeTags, value::Value> ByteCode::aggDoubleDoubleSumImpl(
+    value::TypeTags lhsTag, value::Value lhsValue, value::TypeTags rhsTag, value::Value rhsValue) {
+    if (!isNumber(rhsTag)) {
+        auto [tag, val] = value::copyValue(lhsTag, lhsValue);
+        return {true, tag, val};
+    }
+
+    tassert(5755317, "The result slot must be Array-typed", lhsTag == TypeTags::Array);
+    auto arr = getArrayView(lhsValue);
+    tassert(5755310,
+            str::stream() << "The result slot must have at least "
+                          << AggSumValueElems::kMaxSizeOfArray - 1
+                          << " elements but got: " << arr->size(),
+            arr->size() >= AggSumValueElems::kMaxSizeOfArray - 1);
+
+    // Only uses tag information from the kNonDecimalTotalTag element.
+    auto [nonDecimalTotalTag, _] = arr->getAt(AggSumValueElems::kNonDecimalTotalTag);
+    tassert(5755311,
+            "The nonDecimalTag can't be NumberDecimal",
+            nonDecimalTotalTag != TypeTags::NumberDecimal);
+    // Only uses values from the kNonDecimalTotalSum/kNonDecimalTotalAddend elements.
+    auto [sumTag, sum] = arr->getAt(AggSumValueElems::kNonDecimalTotalSum);
+    auto [addendTag, addend] = arr->getAt(AggSumValueElems::kNonDecimalTotalAddend);
+    tassert(5755312,
+            "The sum and addend must be NumberDouble",
+            sumTag == addendTag && sumTag == TypeTags::NumberDouble);
+
+    // We're guaranteed to always have a valid nonDecimalTotal value.
+    auto nonDecimalTotal = DoubleDoubleSummation::create(value::bitcastTo<double>(sum),
+                                                         value::bitcastTo<double>(addend));
+
+    if (auto nElems = arr->size(); nElems < AggSumValueElems::kMaxSizeOfArray) {
+        // We haven't seen any decimal value so far.
+        if (auto totalTag = getWidestNumericalType(nonDecimalTotalTag, rhsTag);
+            totalTag == TypeTags::NumberDecimal) {
+            // We have seen a decimal for the first time and start storing the total sum
+            // of decimal values into 'kDecimalTotal' element and the total sum of non-decimal
+            // values into 'kNonDecimalXXX' elements.
+            tassert(
+                5755313, "The arg value must be NumberDecimal", rhsTag == TypeTags::NumberDecimal);
+
+            return makeSumResultArray(
+                nonDecimalTotalTag, nonDecimalTotal, value::bitcastTo<Decimal128>(rhsValue));
+        } else {
+            addNonDecimal(rhsTag, rhsValue, nonDecimalTotal);
+            return makeSumResultArray(totalTag, nonDecimalTotal);
+        }
+    } else {
+        // We've seen a decimal value. We've already started storing the total sum of decimal values
+        // into 'kDecimalTotal' element and the total sum of non-decimal values into
+        // 'kNonDecimalXXX' elements.
+        tassert(5755314,
+                str::stream() << "The result slot must have at most "
+                              << AggSumValueElems::kMaxSizeOfArray
+                              << " elements but got: " << arr->size(),
+                nElems == AggSumValueElems::kMaxSizeOfArray);
+        auto [decimalTotalTag, decimalTotalVal] = arr->getAt(AggSumValueElems::kDecimalTotal);
+        tassert(5755315,
+                "The decimalTotal must be NumberDecimal",
+                decimalTotalTag == TypeTags::NumberDecimal);
+
+        auto decimalTotal = value::bitcastTo<Decimal128>(decimalTotalVal);
+        if (rhsTag == TypeTags::NumberDecimal) {
+            decimalTotal = decimalTotal.add(value::bitcastTo<Decimal128>(rhsValue));
+        } else {
+            nonDecimalTotalTag = getWidestNumericalType(nonDecimalTotalTag, rhsTag);
+            addNonDecimal(rhsTag, rhsValue, nonDecimalTotal);
+        }
+
+        return makeSumResultArray(nonDecimalTotalTag, nonDecimalTotal, decimalTotal);
+    }
+}
+
 std::tuple<bool, value::TypeTags, value::Value> ByteCode::genericSub(value::TypeTags lhsTag,
                                                                      value::Value lhsValue,
                                                                      value::TypeTags rhsTag,
diff --git a/src/mongo/db/exec/sbe/vm/vm.cpp b/src/mongo/db/exec/sbe/vm/vm.cpp
index 57687160f24..ea9695c63b4 100644
--- a/src/mongo/db/exec/sbe/vm/vm.cpp
+++ b/src/mongo/db/exec/sbe/vm/vm.cpp
@@ -112,6 +112,8 @@ int Instruction::stackOffset[Instruction::Tags::lastInstruction] = {
     0,   // getArraySize
 
     -1,  // aggSum
+    -1,  // aggDoubleDoubleSum
+    0,   // doubleDoubleSumFinalize
     -1,  // aggMin
     -1,  // aggMax
     -1,  // aggFirst
@@ -364,6 +366,14 @@ void CodeFragment::appendSum() {
     appendSimpleInstruction(Instruction::aggSum);
 }
 
+void CodeFragment::appendDoubleDoubleSum() {
+    appendSimpleInstruction(Instruction::aggDoubleDoubleSum);
+}
+
+void CodeFragment::appendDoubleDoubleSumFinalize() {
+    appendSimpleInstruction(Instruction::doubleDoubleSumFinalize);
+}
+
 void CodeFragment::appendMin() {
     appendSimpleInstruction(Instruction::aggMin);
 }
@@ -899,6 +909,103 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::aggSum(value::TypeTags
     return genericAdd(accTag, accValue, fieldTag, fieldValue);
 }
 
+std::tuple<bool, value::TypeTags, value::Value> ByteCode::aggDoubleDoubleSum(
+    value::TypeTags accTag,
+    value::Value accValue,
+    value::TypeTags fieldTag,
+    value::Value fieldValue) {
+    // Skip aggregation step if we don't have the input.
+    if (fieldTag == value::TypeTags::Nothing) {
+        auto [tag, val] = value::copyValue(accTag, accValue);
+        return {true, tag, val};
+    }
+
+    // Initialize the accumulator.
+    if (accTag == value::TypeTags::Nothing) {
+        auto [accTagN, accValueN] = value::makeNewArray();
+        value::ValueGuard guard{accTagN, accValueN};
+        auto arr = value::getArrayView(accValueN);
+
+        // The order of the following three elements should match to 'AggSumValueElems'.
+        arr->push_back(value::TypeTags::NumberInt32, value::bitcastFrom<int32_t>(0));
+        arr->push_back(value::TypeTags::NumberDouble, value::bitcastFrom<double>(0.0));
+        arr->push_back(value::TypeTags::NumberDouble, value::bitcastFrom<double>(0.0));
+        // The absent 'kDecimalTotal' element means that we've not seen any decimal value. So, we're
+        // not adding 'kDecimalTotal' element yet.
+        return aggDoubleDoubleSumImpl(accTagN, accValueN, fieldTag, fieldValue);
+    }
+
+    return aggDoubleDoubleSumImpl(accTag, accValue, fieldTag, fieldValue);
+}
+
+std::tuple<bool, value::TypeTags, value::Value> ByteCode::doubleDoubleSumFinalize(
+    value::TypeTags fieldTag, value::Value fieldValue) {
+    auto arr = value::getArrayView(fieldValue);
+    tassert(5755321,
+            str::stream() << "The result slot must have at least "
+                          << AggSumValueElems::kMaxSizeOfArray - 1
+                          << " elements but got: " << arr->size(),
+            arr->size() >= AggSumValueElems::kMaxSizeOfArray - 1);
+
+    auto nonDecimalTotalTag = arr->getAt(AggSumValueElems::kNonDecimalTotalTag).first;
+    tassert(5755322,
+            "The nonDecimalTag can't be NumberDecimal",
+            nonDecimalTotalTag != value::TypeTags::NumberDecimal);
+    auto [sumTag, sum] = arr->getAt(AggSumValueElems::kNonDecimalTotalSum);
+    auto [addendTag, addend] = arr->getAt(AggSumValueElems::kNonDecimalTotalAddend);
+    tassert(5755323,
+            "The sum and addend must be NumbetDouble",
+            sumTag == addendTag && sumTag == value::TypeTags::NumberDouble);
+
+    // We're guaranteed to always have a valid nonDecimalTotal value.
+    auto nonDecimalTotal = DoubleDoubleSummation::create(value::bitcastTo<double>(sum),
+                                                         value::bitcastTo<double>(addend));
+
+    if (auto nElems = arr->size(); nElems < AggSumValueElems::kMaxSizeOfArray) {
+        // We've not seen any decimal value.
+        switch (nonDecimalTotalTag) {
+            case value::TypeTags::NumberInt32:
+            case value::TypeTags::NumberInt64:
+                if (nonDecimalTotal.fitsLong()) {
+                    auto longVal = nonDecimalTotal.getLong();
+                    if (int intVal = longVal;
+                        nonDecimalTotalTag == value::TypeTags::NumberInt32 && intVal == longVal) {
+                        return {true,
+                                value::TypeTags::NumberInt32,
+                                value::bitcastFrom<int32_t>(intVal)};
+                    } else {
+                        return {true,
+                                value::TypeTags::NumberInt64,
+                                value::bitcastFrom<int64_t>(longVal)};
+                    }
+                }
+                // Sum doesn't fit a NumberLong, so return a NumberDouble instead.
+                [[fallthrough]];
+            case value::TypeTags::NumberDouble:
+                return {true,
+                        value::TypeTags::NumberDouble,
+                        value::bitcastFrom<double>(nonDecimalTotal.getDouble())};
+            default:
+                MONGO_UNREACHABLE_TASSERT(5755324);
+        }
+    } else {
+        // We've seen a decimal value.
+        tassert(5755325,
+                str::stream() << "The result slot must have at most "
+                              << AggSumValueElems::kMaxSizeOfArray
+                              << " elements but got: " << arr->size(),
+                nElems == AggSumValueElems::kMaxSizeOfArray);
+        auto [decimalTotalTag, decimalTotalVal] = arr->getAt(AggSumValueElems::kDecimalTotal);
+        tassert(5755326,
+                "The decimalTotal must be NumberDecimal",
+                decimalTotalTag == value::TypeTags::NumberDecimal);
+
+        auto decimalTotal = value::bitcastTo<Decimal128>(decimalTotalVal);
+        auto [tag, val] = value::makeCopyDecimal(decimalTotal.add(nonDecimalTotal.getDecimal()));
+        return {true, tag, val};
+    }
+}
+
 std::tuple<bool, value::TypeTags, value::Value> ByteCode::aggMin(value::TypeTags accTag,
                                                                  value::Value accValue,
                                                                  value::TypeTags fieldTag,
@@ -4381,6 +4488,35 @@ void ByteCode::runInternal(const CodeFragment* code, int64_t position) {
                     }
                     break;
                 }
+                case Instruction::aggDoubleDoubleSum: {
+                    auto [rhsOwned, rhsTag, rhsVal] = getFromStack(0);
+                    popStack();
+                    auto [lhsOwned, lhsTag, lhsVal] = getFromStack(0);
+
+                    auto [owned, tag, val] = aggDoubleDoubleSum(lhsTag, lhsVal, rhsTag, rhsVal);
+
+                    topStack(owned, tag, val);
+
+                    if (rhsOwned) {
+                        value::releaseValue(rhsTag, rhsVal);
+                    }
+                    if (lhsOwned) {
+                        value::releaseValue(lhsTag, lhsVal);
+                    }
+                    break;
+                }
+                case Instruction::doubleDoubleSumFinalize: {
+                    auto [sumArrayOwned, sumArrayTag, sumArrayVal] = getFromStack(0);
+                    auto [finalSumOwned, finalSumTag, finalSumVal] =
+                        doubleDoubleSumFinalize(sumArrayTag, sumArrayVal);
+
+                    topStack(finalSumOwned, finalSumTag, finalSumVal);
+
+                    if (sumArrayOwned) {
+                        value::releaseValue(sumArrayTag, sumArrayVal);
+                    }
+                    break;
+                }
                 case Instruction::aggMin: {
                     auto [rhsOwned, rhsTag, rhsVal] = getFromStack(0);
                     popStack();
diff --git a/src/mongo/db/exec/sbe/vm/vm.h b/src/mongo/db/exec/sbe/vm/vm.h
index 8f7c3525ea2..776e6487765 100644
--- a/src/mongo/db/exec/sbe/vm/vm.h
+++ b/src/mongo/db/exec/sbe/vm/vm.h
@@ -249,6 +249,8 @@ struct Instruction {
         getArraySize,
 
         aggSum,
+        aggDoubleDoubleSum,
+        doubleDoubleSumFinalize,
         aggMin,
         aggMax,
         aggFirst,
@@ -381,6 +383,29 @@ enum class Builtin : uint8_t {
     tsIncrement,
 };
 
+/**
+ * This enum defines indices into an 'Array' that accumulates $sum results.
+ *
+ * The array might contain up to four elements:
+ * - The element at index `kNonDecimalTotalTag` keeps track of the widest type of the sum of
+ * non-decimal values.
+ * - The elements at indices `kNonDecimalTotalSum` and `kNonDecimalTotalAddend` together represent
+ * non-decimal value which is the sum of all non-decimal values.
+ * - The element at index `kDecimalTotal` is optional and represents the sum of all decimal values
+ * if any such values are encountered.
+ *
+ * See 'aggDoubleDoubleSumImpl()'/'aggDoubleDoubleSum()'/'doubleDoubleSumFinalize()' for more
+ * details.
+ */
+enum AggSumValueElems {
+    kNonDecimalTotalTag,
+    kNonDecimalTotalSum,
+    kNonDecimalTotalAddend,
+    kDecimalTotal,
+    // This is actually not an index but represents the maximum number of elements.
+    kMaxSizeOfArray
+};
+
 using SmallArityType = uint8_t;
 using ArityType = uint32_t;
 
@@ -480,6 +505,8 @@ public:
     void appendGetArraySize();
 
     void appendSum();
+    void appendDoubleDoubleSum();
+    void appendDoubleDoubleSumFinalize();
     void appendMin();
     void appendMax();
     void appendFirst();
@@ -667,6 +694,21 @@ private:
                                                            value::TypeTags fieldTag,
                                                            value::Value fieldValue);
 
+    std::tuple<bool, value::TypeTags, value::Value> aggDoubleDoubleSumImpl(value::TypeTags lhsTag,
+                                                                           value::Value lhsValue,
+                                                                           value::TypeTags rhsTag,
+                                                                           value::Value rhsValue);
+
+    std::tuple<bool, value::TypeTags, value::Value> aggDoubleDoubleSum(value::TypeTags accTag,
+                                                                       value::Value accValue,
+                                                                       value::TypeTags fieldTag,
+                                                                       value::Value fieldValue);
+
+    // This function is necessary because 'aggDoubleDoubleSum()' result is 'Array' type but we need
+    // to produce a scalar value out of it.
+    std::tuple<bool, value::TypeTags, value::Value> doubleDoubleSumFinalize(
+        value::TypeTags fieldTag, value::Value fieldValue);
+
     std::tuple<bool, value::TypeTags, value::Value> aggMin(value::TypeTags accTag,
                                                            value::Value accValue,
                                                            value::TypeTags fieldTag,
diff --git a/src/mongo/db/query/sbe_stage_builder_accumulator.cpp b/src/mongo/db/query/sbe_stage_builder_accumulator.cpp
index 81af5232152..cf6abe6d8e6 100644
--- a/src/mongo/db/query/sbe_stage_builder_accumulator.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_accumulator.cpp
@@ -170,6 +170,34 @@ std::pair<std::unique_ptr<sbe::EExpression>, EvalStage> buildFinalizeAvg(
         std::move(inputStage)};
 }
 
+std::pair<std::vector<std::unique_ptr<sbe::EExpression>>, EvalStage> buildAccumulatorSum(
+    StageBuilderState& state,
+    const AccumulationExpression& expr,
+    std::unique_ptr<sbe::EExpression> arg,
+    EvalStage inputStage,
+    PlanNodeId planNodeId) {
+    std::vector<std::unique_ptr<sbe::EExpression>> aggs;
+    aggs.push_back(makeFunction("aggDoubleDoubleSum", std::move(arg)));
+    return {std::move(aggs), std::move(inputStage)};
+}
+
+std::pair<std::unique_ptr<sbe::EExpression>, EvalStage> buildFinalizeSum(
+    StageBuilderState& state,
+    const AccumulationExpression& expr,
+    const sbe::value::SlotVector& sumSlots,
+    EvalStage inputStage,
+    PlanNodeId planNodeId) {
+    tassert(5755300,
+            str::stream() << "Expected one input slot for finalization of sum, got: "
+                          << sumSlots.size(),
+            sumSlots.size() == 1);
+    auto sumFinalize =
+        sbe::makeE<sbe::EIf>(generateNullOrMissing(makeVariable(sumSlots[0])),
+                             makeConstant(sbe::value::TypeTags::NumberInt32, 0),
+                             makeFunction("doubleDoubleSumFinalize", makeVariable(sumSlots[0])));
+    return {std::move(sumFinalize), std::move(inputStage)};
+}
+
 std::pair<std::vector<std::unique_ptr<sbe::EExpression>>, EvalStage> buildAccumulatorAddToSet(
     StageBuilderState& state,
     const AccumulationExpression& expr,
@@ -228,6 +256,7 @@ std::pair<std::vector<std::unique_ptr<sbe::EExpression>>, EvalStage> buildAccumu
         {AccumulatorFirst::kName, &buildAccumulatorFirst},
         {AccumulatorLast::kName, &buildAccumulatorLast},
         {AccumulatorAvg::kName, &buildAccumulatorAvg},
+        {AccumulatorSum::kName, &buildAccumulatorSum},
         {AccumulatorAddToSet::kName, &buildAccumulatorAddToSet}};
 
     auto accExprName = acc.expr.name;
@@ -262,6 +291,7 @@ std::pair<std::unique_ptr<sbe::EExpression>, EvalStage> buildFinalize(
         {AccumulatorFirst::kName, &buildFinalizeFirst},
         {AccumulatorLast::kName, &buildFinalizeLast},
         {AccumulatorAvg::kName, &buildFinalizeAvg},
+        {AccumulatorSum::kName, &buildFinalizeSum},
         {AccumulatorAddToSet::kName, &buildFinalizeAddToSet}};
 
     auto accExprName = acc.expr.name;
diff --git a/src/mongo/db/query/sbe_stage_builder_accumulator_test.cpp b/src/mongo/db/query/sbe_stage_builder_accumulator_test.cpp
index 3d2de6e6687..1f5b08af983 100644
--- a/src/mongo/db/query/sbe_stage_builder_accumulator_test.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_accumulator_test.cpp
@@ -135,7 +135,6 @@ protected:
         auto [resultsTag, resultsVal] =
             getResultsForAggregationWithNoGroupByTest(queryStatement, docs);
         sbe::value::ValueGuard resultGuard{resultsTag, resultsVal};
-
         auto [expectedTag, expectedVal] = stage_builder::makeValue(expectedValue);
         sbe::value::ValueGuard expectedGuard{expectedTag, expectedVal};
         ASSERT_TRUE(valueEquals(resultsTag, resultsVal, expectedTag, expectedVal));
@@ -704,6 +703,275 @@ TEST_F(SbeAccumulatorBuilderTest, AvgAccumulatorOneGroupByTranslation) {
     ASSERT_TRUE(valueEquals(sortedResultsTag, sortedResultsVal, expectedTag, expectedVal));
 }
 
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationBasic) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 2)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 4)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 6))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(12));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationOneDocInt) {
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}", {BSON_ARRAY(BSON("a" << 1 << "b" << 10))}, BSON_ARRAY(Value(10)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationOneDocLong) {
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 3 << "b" << 10ll))},
+        BSON_ARRAY(10ll));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationOneDocIntRepresentableDouble) {
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 4 << "b" << 10.0))},
+        BSON_ARRAY(10.0));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationOneDocNonIntRepresentableLong) {
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 5 << "b" << 60000000000ll))},
+        BSON_ARRAY(60000000000ll));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationNonIntegerValuedDouble) {
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 9.5))},
+        BSON_ARRAY(9.5));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationOneDocNanDouble) {
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        std::vector<BSONArray>{
+            BSON_ARRAY(BSON("a" << 6 << "b" << std::numeric_limits<double>::quiet_NaN()))},
+        BSON_ARRAY(Value(std::numeric_limits<double>::quiet_NaN())));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntAndLong) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 4)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 5ll))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(Value(9ll)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntAndDouble) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 4)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 5.5))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(9.5));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationLongAndDouble) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 4ll)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 5.5))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(9.5));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntLongDouble) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 4)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 4ll)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 5.5))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(13.5));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationLongLongDecimal) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 4ll)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 4ll)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << Decimal128("5.5")))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(Decimal128("13.5")));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationDoubleAndDecimal) {
+    const auto doubleVal = 4.2;
+    DoubleDoubleSummation doubleDoubleSum;
+    doubleDoubleSum.addDouble(doubleVal);
+    const auto nonDecimal = doubleDoubleSum.getDecimal();
+
+    const auto decimalVal = Decimal128("5.5");
+    const auto expected = decimalVal.add(nonDecimal);
+
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << doubleVal)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << decimalVal))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(expected));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntAndMissing) {
+    auto docs =
+        std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 4)), BSON_ARRAY(BSON("a" << 1))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(4));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationMixedTypesWithDecimal128) {
+    const auto doubleVal1 = 4.8;
+    const auto doubleVal2 = -5.0;
+    const auto intVal = 6;
+    const auto longVal = 3l;
+
+    DoubleDoubleSummation doubleDoubleSum;
+    doubleDoubleSum.addDouble(doubleVal1);
+    doubleDoubleSum.addInt(intVal);
+    doubleDoubleSum.addInt(longVal);
+    doubleDoubleSum.addDouble(doubleVal2);
+
+    const auto decimalVal = Decimal128(1.0);
+    const auto expected = decimalVal.add(doubleDoubleSum.getDecimal());
+
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << BSON_ARRAY(2 << 4 << 6))),
+                                       BSON_ARRAY(BSON("a" << 2 << "b" << doubleVal1)),
+                                       BSON_ARRAY(BSON("a" << 2)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << intVal)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << longVal)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << decimalVal)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << doubleVal2)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << BSONNULL)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << BSON("c" << 1)))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(expected));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationDecimalSum) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << Decimal128("-10.100"))),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << Decimal128("20.200")))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(Decimal128("10.100")));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationAllNull) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << BSONNULL)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << BSONNULL)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << BSONNULL))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(0));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationAllMissing) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1)), BSON_ARRAY(BSON("a" << 1))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(0));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationSomeNonNumeric) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b"
+                                                           << "c")),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 3)),
+                                       BSON_ARRAY(BSON("a" << 1 << "b"
+                                                           << "m")),
+                                       BSON_ARRAY(BSON("a" << 1 << "b" << 4))};
+    runAggregationWithNoGroupByTest("{x: {$sum: '$b'}}", docs, BSON_ARRAY(7));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationTwoIntsDoNotOverflow) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<int>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 10))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}", docs, BSON_ARRAY(Value(std::numeric_limits<int>::max() + 10LL)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationTwoNegativeIntsDoNotOverflow) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(-std::numeric_limits<int>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << -10))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}", docs, BSON_ARRAY(Value(-std::numeric_limits<int>::max() - 10LL)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntAndLongDoNotTriggerIntOverflow) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<int>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 1LL))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        docs,
+        BSON_ARRAY(Value(static_cast<long long>(std::numeric_limits<int>::max()) + 1)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntAndDoubleDoNotTriggerOverflow) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<int>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 1.0))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        docs,
+        BSON_ARRAY(Value(static_cast<long long>(std::numeric_limits<int>::max()) + 1.0)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntAndLongOverflowIntoDouble) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<long long>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 1))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        docs,
+        BSON_ARRAY(Value(-static_cast<double>(std::numeric_limits<long long>::min()))));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationTwoLongsOverflowIntoDouble) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<long long>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<long long>::max())))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        docs,
+        BSON_ARRAY(Value(static_cast<double>(std::numeric_limits<long long>::max()) * 2)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationLongAndDoubleDoNotTriggerLongOverflow) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<long long>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 1.0))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}", docs, BSON_ARRAY(Value(std::numeric_limits<long long>::max() + 1.0)));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationTwoDoublesOverflowToInfinity) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<double>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<double>::max())))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}", docs, BSON_ARRAY(Value(std::numeric_limits<double>::infinity())));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationTwoIntegersDoNotOverflowIfDoubleAdded) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<long long>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<long long>::max()))),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 1.0))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}",
+        docs,
+        BSON_ARRAY(Value(static_cast<double>(std::numeric_limits<long long>::max()) +
+                         static_cast<double>(std::numeric_limits<long long>::max()))));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationIntAndNanDouble) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << 3)),
+        BSON_ARRAY(BSON("a" << 1 << "b" << Value(std::numeric_limits<double>::quiet_NaN())))};
+    runAggregationWithNoGroupByTest(
+        "{x: {$sum: '$b'}}", docs, BSON_ARRAY(Value(std::numeric_limits<double>::quiet_NaN())));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationGroupByTest) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << 3)),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 4)),
+        BSON_ARRAY(BSON("a" << 2 << "b" << 13)),
+        BSON_ARRAY(BSON("a" << 2 << "b"
+                            << "a")),
+    };
+    runAggregationWithGroupByTest("{x: {$sum: '$b'}}", docs, {"$a"}, BSON_ARRAY(7 << 13));
+}
+
+TEST_F(SbeAccumulatorBuilderTest, SumAccumulatorTranslationTwoGroupByTest) {
+    auto docs = std::vector<BSONArray>{
+        BSON_ARRAY(BSON("a" << 1 << "b" << 3 << "c" << 1)),
+        BSON_ARRAY(BSON("a" << 1 << "b" << 4 << "c" << 1)),
+        BSON_ARRAY(BSON("a" << 2 << "b" << 13 << "c" << 1)),
+        BSON_ARRAY(BSON("a" << 2 << "b"
+                            << "a"
+                            << "c" << 1)),
+    };
+    runAggregationWithGroupByTest("{x: {$sum: '$b'}}", docs, {"$a", "$c"}, BSON_ARRAY(20));
+}
 TEST_F(SbeAccumulatorBuilderTest, AddToSetAccumulatorTranslationSingleDoc) {
     auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 1))};
     runAddToSetTest("{x: {$addToSet: '$b'}}", docs, BSON_ARRAY(1));
diff --git a/src/mongo/util/summation.h b/src/mongo/util/summation.h
index d199fc304a7..18f03e35151 100644
--- a/src/mongo/util/summation.h
+++ b/src/mongo/util/summation.h
@@ -48,6 +48,15 @@ using DoubleDouble = std::pair<double, double>;
  */
 class DoubleDoubleSummation {
 public:
+    DoubleDoubleSummation() = default;
+
+    /**
+     * Factory method.
+     */
+    static constexpr DoubleDoubleSummation create(double sum, double addend) noexcept {
+        return DoubleDoubleSummation(sum, addend);
+    }
+
     /**
      * Adds x to the sum, keeping track of a compensation amount to be subtracted later.
      */
@@ -151,5 +160,8 @@ private:
     // Simple sum to be returned if _sum is NaN. This addresses infinities turning into NaNs when
     // using compensated addition.
     double _special = 0.0;
+
+    constexpr DoubleDoubleSummation(double sum, double addend) noexcept
+        : _sum(sum), _addend(addend), _special(sum) {}
 };
 }  // namespace mongo