SERVER-19624 Add Decimal128 type support to mongo/bson layer

19 files changed, 703 insertions, 28 deletions

diff --git a/src/mongo/base/compare_numbers.h b/src/mongo/base/compare_numbers.h
index 0049d5ba336..c58d91a3a34 100644
--- a/src/mongo/base/compare_numbers.h
+++ b/src/mongo/base/compare_numbers.h
@@ -29,6 +29,7 @@
 
 #include <cmath>
 
+#include "mongo/platform/decimal128.h"
 #include "mongo/util/assert_util.h"
 
 namespace mongo {
@@ -37,6 +38,11 @@ namespace mongo {
  * These functions compare numbers using the same rules as BSON. Care is taken to always give
  * numerically correct results when comparing different types. Returns are always -1, 0, or 1 to
  * ensure it is safe to negate the result to invert the direction of the comparison.
+ *
+ * lhs > rhs returns 1
+ * lhs < rhs returns -1
+ * lhs == rhs returns 0
+ *
  */
 
 inline int compareInts(int lhs, int rhs) {
@@ -98,4 +104,76 @@ inline int compareDoubleToLong(double lhs, long long rhs) {
     return -compareLongToDouble(rhs, lhs);
 }
 
+/** Decimal type comparisons
+ * These following cases need support:
+ * 1. decimal and decimal: directly compare (enforce ordering: NaN < -Inf < N < +Inf)
+ * 2. decimal and int: convert int to decimal and compare
+ * 3. decimal and long: convert long to decimal and compare
+ * 4. decimal to double: convert decimal to double with round toward negative.
+ *    Check for exact conversion and determine ordering based on result.
+ */
+
+// Case 1: Compare two decimal values, but enforce MongoDB's total ordering convention
+inline int compareDecimals(Decimal128 lhs, Decimal128 rhs) {
+    // When we're comparing, lhs is always a decimal, which means more often than not
+    // the rhs will be less than the lhs (decimal type has the largest capacity)
+    if (lhs.isGreater(rhs))
+        return 1;
+    if (lhs.isLess(rhs))
+        return -1;
+    if (lhs.isNaN())
+        return (rhs.isNaN() ? 0 : -1);
+    if (rhs.isNaN())
+        return 1;
+    else  // lhs is necessarily equal to rhs
+        return 0;
+}
+
+// Compare decimal and int
+inline int compareDecimalToInt(Decimal128 lhs, int rhs) {
+    return compareDecimals(lhs, Decimal128(rhs));
+}
+
+inline int compareIntToDecimal(int lhs, Decimal128 rhs) {
+    return -compareDecimalToInt(rhs, lhs);
+}
+
+// Compare decimal and long
+inline int compareDecimalToLong(Decimal128 lhs, long long rhs) {
+    return compareDecimals(lhs, Decimal128(rhs));
+}
+
+inline int compareLongToDecimal(long long lhs, Decimal128 rhs) {
+    return -compareDecimalToLong(rhs, lhs);
+}
+
+// Compare decimal and double
+inline int compareDecimalToDouble(Decimal128 lhs, double rhs) {
+    uint32_t sigFlags = Decimal128::SignalingFlag::kNoFlag;
+    double decToDouble = lhs.toDouble(&sigFlags, Decimal128::RoundingMode::kRoundTowardNegative);
+
+    if (decToDouble == rhs) {
+        // If our conversion was not exact, lhs was necessarily greater than rhs
+        // otherwise, they are equal
+        if (Decimal128::hasFlag(sigFlags, Decimal128::SignalingFlag::kInexact)) {
+            return 1;
+        } else {
+            return 0;
+        }
+    } else if (decToDouble < rhs) {
+        return -1;
+    } else if (decToDouble > rhs) {
+        return 1;
+    }
+
+    if (lhs.isNaN())
+        return (std::isnan(rhs) ? 0 : -1);
+    invariant(std::isnan(rhs));
+    return 1;
+}
+
+inline int compareDoubleToDecimal(double lhs, Decimal128 rhs) {
+    return -compareDecimalToDouble(rhs, lhs);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/bson/bson_obj_test.cpp b/src/mongo/bson/bson_obj_test.cpp
index 6a80deb3609..07735eb6025 100644
--- a/src/mongo/bson/bson_obj_test.cpp
+++ b/src/mongo/bson/bson_obj_test.cpp
@@ -27,6 +27,7 @@
 
 #include "mongo/db/jsobj.h"
 #include "mongo/db/json.h"
+#include "mongo/platform/decimal128.h"
 
 #include "mongo/unittest/unittest.h"
 
@@ -183,6 +184,206 @@ TEST(BSONObjCompare, NumberLong_Double) {
     }
 }
 
+TEST(BSONObjCompare, NumberDecimalScaleAndZero) {
+    if (Decimal128::enabled) {
+        ASSERT_LT(BSON("" << Decimal128(0.0)), BSON("" << Decimal128(1.0)));
+        ASSERT_LT(BSON("" << Decimal128(-1.0)), BSON("" << Decimal128(0.0)));
+        ASSERT_LT(BSON("" << Decimal128(-1.0)), BSON("" << Decimal128(1.0)));
+
+        ASSERT_LT(BSON("" << Decimal128(0.0)), BSON("" << Decimal128(0.1)));
+        ASSERT_LT(BSON("" << Decimal128(0.1)), BSON("" << Decimal128(1.0)));
+        ASSERT_LT(BSON("" << Decimal128(-1.0)), BSON("" << Decimal128(-0.1)));
+        ASSERT_LT(BSON("" << Decimal128(-0.1)), BSON("" << Decimal128(-0.0)));
+        ASSERT_LT(BSON("" << Decimal128(-0.1)), BSON("" << Decimal128(0.1)));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalMaxAndMins) {
+    if (Decimal128::enabled) {
+        ASSERT_LT(BSON("" << Decimal128(0.0)), BSON("" << Decimal128::kSmallestPositive));
+        ASSERT_GT(BSON("" << Decimal128(0.0)), BSON("" << Decimal128::kLargestNegative));
+
+        // over 34 digits of precision so it should be equal
+        ASSERT_EQ(BSON("" << Decimal128(1.0)),
+                  BSON("" << Decimal128(1.0).add(Decimal128::kSmallestPositive)));
+        ASSERT_EQ(BSON("" << Decimal128(0.0)), BSON("" << Decimal128(-0.0)));
+
+        ASSERT_EQ(BSON("" << Decimal128(0)), BSON("" << Decimal128(0)));
+        ASSERT_EQ(BSON("" << Decimal128::kSmallestPositive),
+                  BSON("" << Decimal128::kSmallestPositive));
+        ASSERT_EQ(BSON("" << Decimal128::kLargestNegative),
+                  BSON("" << Decimal128::kLargestNegative));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalInfinity) {
+    if (Decimal128::enabled) {
+        ASSERT_GT(BSON("" << Decimal128::kPositiveInfinity), BSON("" << Decimal128(0.0)));
+        ASSERT_GT(BSON("" << Decimal128::kPositiveInfinity),
+                  BSON("" << Decimal128::kLargestPositive));
+        ASSERT_GT(BSON("" << Decimal128::kPositiveInfinity),
+                  BSON("" << Decimal128::kNegativeInfinity));
+
+        ASSERT_EQ(BSON("" << Decimal128::kPositiveInfinity),
+                  BSON("" << Decimal128::kPositiveInfinity));
+        ASSERT_EQ(BSON("" << Decimal128::kNegativeInfinity),
+                  BSON("" << Decimal128::kNegativeInfinity));
+
+        ASSERT_LT(BSON("" << Decimal128::kNegativeInfinity), BSON("" << Decimal128(0.0)));
+        ASSERT_LT(BSON("" << Decimal128::kNegativeInfinity),
+                  BSON("" << Decimal128::kSmallestNegative));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalPosNaN) {
+    if (Decimal128::enabled) {
+        // +/-NaN is well ordered and compares smallest, so +NaN and -NaN should behave the same
+        ASSERT_LT(BSON("" << Decimal128::kPositiveNaN), BSON("" << 0.0));
+        ASSERT_LT(BSON("" << Decimal128::kPositiveNaN), BSON("" << Decimal128::kSmallestNegative));
+        ASSERT_LT(BSON("" << Decimal128::kPositiveNaN), BSON("" << Decimal128::kPositiveInfinity));
+        ASSERT_LT(BSON("" << Decimal128::kPositiveNaN), BSON("" << Decimal128::kNegativeInfinity));
+
+        ASSERT_EQ(BSON("" << Decimal128::kPositiveNaN), BSON("" << Decimal128::kNegativeNaN));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalNegNan) {
+    if (Decimal128::enabled) {
+        ASSERT_LT(BSON("" << Decimal128::kNegativeNaN), BSON("" << 0.0));
+        ASSERT_LT(BSON("" << Decimal128::kNegativeNaN), BSON("" << Decimal128::kSmallestNegative));
+        ASSERT_LT(BSON("" << Decimal128::kNegativeNaN), BSON("" << Decimal128::kPositiveInfinity));
+        ASSERT_LT(BSON("" << Decimal128::kNegativeNaN), BSON("" << Decimal128::kNegativeInfinity));
+
+        ASSERT_EQ(BSON("" << Decimal128::kNegativeNaN), BSON("" << Decimal128::kPositiveNaN));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalCompareInt) {
+    if (Decimal128::enabled) {
+        ASSERT_EQ(BSON("" << Decimal128(0.0)), BSON("" << 0));
+        ASSERT_EQ(BSON("" << Decimal128(502.0)), BSON("" << 502));
+        ASSERT_EQ(BSON("" << Decimal128(std::numeric_limits<int>::max())),
+                  BSON("" << std::numeric_limits<int>::max()));
+        ASSERT_EQ(BSON("" << Decimal128(-std::numeric_limits<int>::max())),
+                  BSON("" << -std::numeric_limits<int>::max()));
+
+        ASSERT_LT(BSON("" << Decimal128::kNegativeNaN),
+                  BSON("" << -std::numeric_limits<int>::max()));
+        ASSERT_LT(BSON("" << Decimal128::kPositiveNaN),
+                  BSON("" << -std::numeric_limits<int>::max()));
+        ASSERT_LT(BSON("" << Decimal128::kNegativeInfinity),
+                  BSON("" << -std::numeric_limits<int>::max()));
+        ASSERT_GT(BSON("" << Decimal128::kPositiveInfinity),
+                  BSON("" << std::numeric_limits<int>::max()));
+
+        ASSERT_GT(BSON("" << Decimal128(1.0)), BSON("" << 0));
+        ASSERT_LT(BSON("" << Decimal128(-1.0)), BSON("" << 0));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalCompareLong) {
+    if (Decimal128::enabled) {
+        ASSERT_EQ(BSON("" << Decimal128(0.0)), BSON("" << 0ll));
+        ASSERT_EQ(BSON("" << Decimal128(502.0)), BSON("" << 502ll));
+        ASSERT_EQ(BSON("" << Decimal128(std::numeric_limits<long long>::max())),
+                  BSON("" << std::numeric_limits<long long>::max()));
+        ASSERT_EQ(BSON("" << Decimal128(-std::numeric_limits<long long>::max())),
+                  BSON("" << -std::numeric_limits<long long>::max()));
+
+        ASSERT_LT(BSON("" << Decimal128::kNegativeNaN),
+                  BSON("" << -std::numeric_limits<long long>::max()));
+        ASSERT_LT(BSON("" << Decimal128::kPositiveNaN),
+                  BSON("" << -std::numeric_limits<long long>::max()));
+        ASSERT_LT(BSON("" << Decimal128::kNegativeInfinity),
+                  BSON("" << -std::numeric_limits<long long>::max()));
+        ASSERT_GT(BSON("" << Decimal128::kPositiveInfinity),
+                  BSON("" << std::numeric_limits<long long>::max()));
+
+        ASSERT_GT(BSON("" << Decimal128(1.0)), BSON("" << 0ll));
+        ASSERT_LT(BSON("" << Decimal128(-1.0)), BSON("" << 0ll));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalCompareDoubleExactRepresentations) {
+    if (Decimal128::enabled) {
+        ASSERT_EQ(BSON("" << Decimal128(0.0)), BSON("" << 0.0));
+        ASSERT_EQ(BSON("" << Decimal128(1.0)), BSON("" << 1.0));
+        ASSERT_EQ(BSON("" << Decimal128(-1.0)), BSON("" << -1.0));
+        ASSERT_EQ(BSON("" << Decimal128(0.125)), BSON("" << 0.125));
+
+        ASSERT_LT(BSON("" << Decimal128(0.0)), BSON("" << 0.125));
+        ASSERT_LT(BSON("" << Decimal128(-1.0)), BSON("" << -0.125));
+
+        ASSERT_GT(BSON("" << Decimal128(1.0)), BSON("" << 0.125));
+        ASSERT_GT(BSON("" << Decimal128(0.0)), BSON("" << -0.125));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalCompareDoubleNoDoubleRepresentation) {
+    if (Decimal128::enabled) {
+        // Double 0.1 should not compare the same as decimal 0.1. The standard
+        // double constructor for decimal types quantizes at 15 places, but this
+        // is not safe for a well ordered comparison because decimal(0.1) would
+        // then compare equal to both double(0.10000000000000000555) and
+        // double(0.999999999999999876). The following test cases check that
+        // proper well ordering is applied to double and decimal comparisons.
+        ASSERT_GT(BSON("" << Decimal128("0.3")), BSON("" << 0.1));
+        ASSERT_LT(BSON("" << Decimal128("0.1")), BSON("" << 0.3));
+        ASSERT_LT(BSON("" << Decimal128("-0.3")), BSON("" << -0.1));
+        ASSERT_GT(BSON("" << Decimal128("-0.1")), BSON("" << -0.3));
+        ASSERT_LT(BSON("" << Decimal128("0.1")), BSON("" << 0.1));
+        ASSERT_GT(BSON("" << Decimal128("0.3")), BSON("" << 0.3));
+        ASSERT_GT(BSON("" << Decimal128("-0.1")), BSON("" << -0.1));
+        ASSERT_LT(BSON("" << Decimal128("-0.3")), BSON("" << -0.3));
+        ASSERT_EQ(BSON("" << Decimal128("0.5")), BSON("" << 0.5));
+        ASSERT_GT(BSON("" << Decimal128("0.5000000000000000000000000000000001")), BSON("" << 0.5));
+
+        // Double 0.1 should compare well against significantly different decimals
+        ASSERT_LT(BSON("" << Decimal128(0.0)), BSON("" << 0.1));
+        ASSERT_GT(BSON("" << Decimal128(1.0)), BSON("" << 0.1));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalCompareDoubleQuantize) {
+    if (Decimal128::enabled) {
+        // These tests deal with doubles that get adjusted when converted to decimal.
+        // The decimal type only will store a double's first 15 decimal digits of
+        // precision (the most it can accurately express).
+        Decimal128 roundedDoubleLargestPosValue("179769313486232E294");
+        Decimal128 roundedDoubleOneAboveLargestPosValue("179769313486233E294");
+        Decimal128 roundedDoubleLargestNegValue("-179769313486232E294");
+        Decimal128 roundedDoubleOneAboveSmallestNegValue("-179769313486231E294");
+
+        ASSERT_EQ(BSON("" << roundedDoubleLargestPosValue),
+                  BSON("" << Decimal128(std::numeric_limits<double>::max())));
+        ASSERT_EQ(BSON("" << roundedDoubleLargestNegValue),
+                  BSON("" << Decimal128(-std::numeric_limits<double>::max())));
+
+        ASSERT_GT(BSON("" << roundedDoubleOneAboveLargestPosValue),
+                  BSON("" << Decimal128(std::numeric_limits<double>::max())));
+        ASSERT_LT(BSON("" << roundedDoubleOneAboveSmallestNegValue),
+                  BSON("" << Decimal128(-std::numeric_limits<double>::min())));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalCompareDoubleInfinity) {
+    if (Decimal128::enabled) {
+        ASSERT_EQ(BSON("" << Decimal128::kPositiveInfinity),
+                  BSON("" << std::numeric_limits<double>::infinity()));
+        ASSERT_EQ(BSON("" << Decimal128::kNegativeInfinity),
+                  BSON("" << -std::numeric_limits<double>::infinity()));
+    }
+}
+
+TEST(BSONObjCompare, NumberDecimalCompareDoubleNaN) {
+    if (Decimal128::enabled) {
+        ASSERT_EQ(BSON("" << Decimal128::kPositiveNaN),
+                  BSON("" << std::numeric_limits<double>::quiet_NaN()));
+        ASSERT_EQ(BSON("" << Decimal128::kNegativeNaN),
+                  BSON("" << -std::numeric_limits<double>::quiet_NaN()));
+    }
+}
+
 TEST(BSONObjCompare, StringSymbol) {
     BSONObj l, r;
     {
diff --git a/src/mongo/bson/bson_validate.cpp b/src/mongo/bson/bson_validate.cpp
index 0b183b271e3..9ef873eb47a 100644
--- a/src/mongo/bson/bson_validate.cpp
+++ b/src/mongo/bson/bson_validate.cpp
@@ -35,6 +35,7 @@
 #include "mongo/bson/bson_validate.h"
 #include "mongo/bson/oid.h"
 #include "mongo/db/jsobj.h"
+#include "mongo/platform/decimal128.h"
 
 namespace mongo {
 
@@ -221,6 +222,17 @@ Status validateElementInfo(Buffer* buffer, ValidationState::State* nextState, BS
                 return makeError("invalid bson", idElem);
             return Status::OK();
 
+        case NumberDecimal:
+            if (Decimal128::enabled) {
+                if (!buffer->skip(sizeof(Decimal128::Value)))
+                    return makeError("Invalid bson", idElem);
+                return Status::OK();
+            } else {
+                return Status(ErrorCodes::InvalidBSON,
+                              "Attempt to use a decimal BSON type when experimental decimal "
+                              "server support is not currently enabled.");
+            }
+
         case DBRef:
             status = buffer->readUTF8String(NULL);
             if (!status.isOK())
diff --git a/src/mongo/bson/bsonelement.cpp b/src/mongo/bson/bsonelement.cpp
index 72be6da4d62..1f8539b2afc 100644
--- a/src/mongo/bson/bsonelement.cpp
+++ b/src/mongo/bson/bsonelement.cpp
@@ -91,6 +91,25 @@ string BSONElement::jsonString(JsonStringFormat format, bool includeFieldNames,
                 massert(10311, message.c_str(), false);
             }
             break;
+        case NumberDecimal:
+            if (format == TenGen)
+                s << "NumberDecimal(\"";
+            else
+                s << "{ \"$numberDecimal\" : \"";
+            // Recognize again that this is not valid JSON according to RFC-4627.
+            // Also, treat -NaN and +NaN as the same thing for MongoDB.
+            if (numberDecimal().isNaN()) {
+                s << "NaN";
+            } else if (numberDecimal().isInfinite()) {
+                s << (numberDecimal().isNegative() ? "-Infinity" : "Infinity");
+            } else {
+                s << numberDecimal().toString();
+            }
+            if (format == TenGen)
+                s << "\")";
+            else
+                s << "\" }";
+            break;
         case mongo::Bool:
             s << (boolean() ? "true" : "false");
             break;
@@ -374,7 +393,6 @@ int BSONElement::woCompare(const BSONElement& e, bool considerFieldName) const {
     return x;
 }
 
-
 BSONObj BSONElement::embeddedObjectUserCheck() const {
     if (MONGO_likely(isABSONObj()))
         return BSONObj(value());
@@ -447,6 +465,9 @@ int BSONElement::size(int maxLen) const {
         case NumberLong:
             x = 8;
             break;
+        case NumberDecimal:
+            x = 16;
+            break;
         case jstOID:
             x = OID::kOIDSize;
             break;
@@ -531,6 +552,9 @@ int BSONElement::size() const {
         case NumberLong:
             x = 8;
             break;
+        case NumberDecimal:
+            x = 16;
+            break;
         case jstOID:
             x = OID::kOIDSize;
             break;
@@ -613,6 +637,9 @@ void BSONElement::toString(StringBuilder& s, bool includeFieldName, bool full, i
         case NumberInt:
             s << _numberInt();
             break;
+        case NumberDecimal:
+            s << _numberDecimal().toString();
+            break;
         case mongo::Bool:
             s << (boolean() ? "true" : "false");
             break;
@@ -743,6 +770,14 @@ bool BSONElement::coerce<double>(double* out) const {
 }
 
 template <>
+bool BSONElement::coerce<Decimal128>(Decimal128* out) const {
+    if (!isNumber())
+        return false;
+    *out = numberDecimal();
+    return true;
+}
+
+template <>
 bool BSONElement::coerce<bool>(bool* out) const {
     *out = trueValue();
     return true;
@@ -854,6 +889,8 @@ int compareElementValues(const BSONElement& l, const BSONElement& r) {
                     return compareLongs(l._numberInt(), r._numberLong());
                 case NumberDouble:
                     return compareDoubles(l._numberInt(), r._numberDouble());
+                case NumberDecimal:
+                    return compareIntToDecimal(l._numberInt(), r._numberDecimal());
                 default:
                     invariant(false);
             }
@@ -867,6 +904,8 @@ int compareElementValues(const BSONElement& l, const BSONElement& r) {
                     return compareLongs(l._numberLong(), r._numberInt());
                 case NumberDouble:
                     return compareLongToDouble(l._numberLong(), r._numberDouble());
+                case NumberDecimal:
+                    return compareLongToDecimal(l._numberLong(), r._numberDecimal());
                 default:
                     invariant(false);
             }
@@ -880,6 +919,23 @@ int compareElementValues(const BSONElement& l, const BSONElement& r) {
                     return compareDoubles(l._numberDouble(), r._numberInt());
                 case NumberLong:
                     return compareDoubleToLong(l._numberDouble(), r._numberLong());
+                case NumberDecimal:
+                    return compareDoubleToDecimal(l._numberDouble(), r._numberDecimal());
+                default:
+                    invariant(false);
+            }
+        }
+
+        case NumberDecimal: {
+            switch (r.type()) {
+                case NumberDecimal:
+                    return compareDecimals(l._numberDecimal(), r._numberDecimal());
+                case NumberInt:
+                    return compareDecimalToInt(l._numberDecimal(), r._numberInt());
+                case NumberLong:
+                    return compareDecimalToLong(l._numberDecimal(), r._numberLong());
+                case NumberDouble:
+                    return compareDecimalToDouble(l._numberDecimal(), r._numberDouble());
                 default:
                     invariant(false);
             }
@@ -972,12 +1028,30 @@ size_t BSONElement::Hasher::operator()(const BSONElement& elem) const {
             boost::hash_combine(hash, elem.date().asInt64());
             break;
 
+        case mongo::NumberDecimal: {
+            const Decimal128 dcml = elem.numberDecimal();
+            if (dcml.toAbs().isGreater(Decimal128(std::numeric_limits<double>::max())) &&
+                !dcml.isInfinite() && !dcml.isNaN()) {
+                // Normalize our decimal to force equivalent decimals
+                // in the same cohort to hash to the same value
+                Decimal128 dcmlNorm(dcml.normalize());
+                boost::hash_combine(hash, dcmlNorm.getValue().low64);
+                boost::hash_combine(hash, dcmlNorm.getValue().high64);
+                break;
+            }
+            // Else, fall through and convert the decimal to a double and hash.
+            // At this point the decimal fits into the range of doubles, is infinity, or is NaN,
+            // which doubles have a cheaper representation for.
+        }
         case mongo::NumberDouble:
         case mongo::NumberLong:
         case mongo::NumberInt: {
             // This converts all numbers to doubles, which ignores the low-order bits of
-            // NumberLongs > 2**53, but that is ok since the hash will still be the same for
-            // equal numbers and is still likely to be different for different numbers.
+            // NumberLongs > 2**53 and precise decimal numbers without double representations,
+            // but that is ok since the hash will still be the same for equal numbers and is
+            // still likely to be different for different numbers. (Note: this issue only
+            // applies for decimals when they are outside of the valid double range. See
+            // the above case.)
             // SERVER-16851
             const double dbl = elem.numberDouble();
             if (std::isnan(dbl)) {
diff --git a/src/mongo/bson/bsonelement.h b/src/mongo/bson/bsonelement.h
index f2fc5416785..fff6ca6928f 100644
--- a/src/mongo/bson/bsonelement.h
+++ b/src/mongo/bson/bsonelement.h
@@ -40,6 +40,8 @@
 #include "mongo/bson/bsontypes.h"
 #include "mongo/bson/oid.h"
 #include "mongo/bson/timestamp.h"
+#include "mongo/config.h"
+#include "mongo/platform/decimal128.h"
 
 namespace mongo {
 class BSONObj;
@@ -54,7 +56,6 @@ typedef BSONObjBuilder bob;
 /* l and r MUST have same type when called: check that first. */
 int compareElementValues(const BSONElement& l, const BSONElement& r);
 
-
 /** BSONElement represents an "element" in a BSONObj.  So for the object { a : 3, b : "abc" },
     'a : 3' is the first element (key+value).
 
@@ -87,6 +88,9 @@ public:
     double Number() const {
         return chk(isNumber()).number();
     }
+    Decimal128 Decimal() const {
+        return chk(NumberDecimal)._numberDecimal();
+    }
     double Double() const {
         return chk(NumberDouble)._numberDouble();
     }
@@ -127,6 +131,9 @@ public:
     void Val(long long& v) const {
         v = Long();
     }
+    void Val(Decimal128& v) const {
+        v = Decimal();
+    }
     void Val(bool& v) const {
         v = Bool();
     }
@@ -289,6 +296,11 @@ public:
         return ConstDataView(value()).read<LittleEndian<int>>();
     }
 
+    /** Return decimal128 value for this field. MUST be NumberDecimal type. */
+    Decimal128 _numberDecimal() const {
+        return Decimal128(ConstDataView(value()).read<LittleEndian<Decimal128::Value>>());
+    }
+
     /** Return long long value for this field. MUST be NumberLong type. */
     long long _numberLong() const {
         return ConstDataView(value()).read<LittleEndian<long long>>();
@@ -308,6 +320,9 @@ public:
      *  very small doubles -> LLONG_MIN  */
     long long safeNumberLong() const;
 
+    /** Retrieve decimal value for the element safely. */
+    Decimal128 numberDecimal() const;
+
     /** Retrieve the numeric value of the element.  If not of a numeric type, returns 0.
         Note: casts to double, data loss may occur with large (>52 bit) NumberLong values.
     */
@@ -630,6 +645,8 @@ inline bool BSONElement::trueValue() const {
             return _numberLong() != 0;
         case NumberDouble:
             return _numberDouble() != 0;
+        case NumberDecimal:
+            return _numberDecimal().isNotEqual(Decimal128(0));
         case NumberInt:
             return _numberInt() != 0;
         case mongo::Bool:
@@ -638,11 +655,9 @@ inline bool BSONElement::trueValue() const {
         case jstNULL:
         case Undefined:
             return false;
-
         default:
-            ;
+            return true;
     }
-    return true;
 }
 
 /** @return true if element is of a numeric type. */
@@ -650,6 +665,9 @@ inline bool BSONElement::isNumber() const {
     switch (type()) {
         case NumberLong:
         case NumberDouble:
+#ifdef MONGO_CONFIG_EXPERIMENTAL_DECIMAL_SUPPORT
+        case NumberDecimal:
+#endif
         case NumberInt:
             return true;
         default:
@@ -662,6 +680,7 @@ inline bool BSONElement::isSimpleType() const {
         case NumberLong:
         case NumberDouble:
         case NumberInt:
+        case NumberDecimal:
         case mongo::String:
         case mongo::Bool:
         case mongo::Date:
@@ -672,6 +691,21 @@ inline bool BSONElement::isSimpleType() const {
     }
 }
 
+inline Decimal128 BSONElement::numberDecimal() const {
+    switch (type()) {
+        case NumberDouble:
+            return Decimal128(_numberDouble());
+        case NumberInt:
+            return Decimal128(_numberInt());
+        case NumberLong:
+            return Decimal128(_numberLong());
+        case NumberDecimal:
+            return _numberDecimal();
+        default:
+            return 0;
+    }
+}
+
 inline double BSONElement::numberDouble() const {
     switch (type()) {
         case NumberDouble:
@@ -680,6 +714,8 @@ inline double BSONElement::numberDouble() const {
             return _numberInt();
         case NumberLong:
             return _numberLong();
+        case NumberDecimal:
+            return _numberDecimal().toDouble();
         default:
             return 0;
     }
@@ -695,6 +731,8 @@ inline int BSONElement::numberInt() const {
             return _numberInt();
         case NumberLong:
             return (int)_numberLong();
+        case NumberDecimal:
+            return _numberDecimal().toInt();
         default:
             return 0;
     }
@@ -709,21 +747,22 @@ inline long long BSONElement::numberLong() const {
             return _numberInt();
         case NumberLong:
             return _numberLong();
+        case NumberDecimal:
+            return _numberDecimal().toLong();
         default:
             return 0;
     }
 }
 
-/** Like numberLong() but with well-defined behavior for doubles that
+/** Like numberLong() but with well-defined behavior for doubles and decimals that
  *  are NaNs, or too large/small to be represented as long longs.
  *  NaNs -> 0
- *  very large doubles -> LLONG_MAX
- *  very small doubles -> LLONG_MIN  */
+ *  very large values -> LLONG_MAX
+ *  very small values -> LLONG_MIN  */
 inline long long BSONElement::safeNumberLong() const {
-    double d;
     switch (type()) {
-        case NumberDouble:
-            d = numberDouble();
+        case NumberDouble: {
+            double d = numberDouble();
             if (std::isnan(d)) {
                 return 0;
             }
@@ -733,6 +772,21 @@ inline long long BSONElement::safeNumberLong() const {
             if (d < std::numeric_limits<long long>::min()) {
                 return std::numeric_limits<long long>::min();
             }
+            return numberLong();
+        }
+        case NumberDecimal: {
+            Decimal128 d = numberDecimal();
+            if (d.isNaN()) {
+                return 0;
+            }
+            if (d.isGreater(Decimal128(std::numeric_limits<long long>::max()))) {
+                return std::numeric_limits<long long>::max();
+            }
+            if (d.isLess(Decimal128(std::numeric_limits<long long>::min()))) {
+                return std::numeric_limits<long long>::min();
+            }
+            return numberLong();
+        }
         default:
             return numberLong();
     }
diff --git a/src/mongo/bson/bsonobj.h b/src/mongo/bson/bsonobj.h
index b7d32dbda5f..28f51e55ec3 100644
--- a/src/mongo/bson/bsonobj.h
+++ b/src/mongo/bson/bsonobj.h
@@ -77,6 +77,7 @@ typedef std::multiset<BSONElement, BSONElementCmpWithoutField> BSONElementMSet;
  OID:       an OID object
  NumberDouble: <double>
  NumberInt: <int32>
+ NumberDecimal: <dec128>
  String:    <unsigned32 strsizewithnull><cstring>
  Date:      <8bytes>
  Regex:     <cstring regex><cstring options>
diff --git a/src/mongo/bson/bsonobjbuilder.cpp b/src/mongo/bson/bsonobjbuilder.cpp
index 59baced7c26..32c46b30a00 100644
--- a/src/mongo/bson/bsonobjbuilder.cpp
+++ b/src/mongo/bson/bsonobjbuilder.cpp
@@ -46,6 +46,7 @@ void BSONObjBuilder::appendMinForType(StringData fieldName, int t) {
         case NumberInt:
         case NumberDouble:
         case NumberLong:
+        case NumberDecimal:
             append(fieldName, std::numeric_limits<double>::quiet_NaN());
             return;
         case Symbol:
@@ -116,6 +117,7 @@ void BSONObjBuilder::appendMaxForType(StringData fieldName, int t) {
         case NumberInt:
         case NumberDouble:
         case NumberLong:
+        case NumberDecimal:
             append(fieldName, std::numeric_limits<double>::infinity());
             return;
         case Symbol:
diff --git a/src/mongo/bson/bsonobjbuilder.h b/src/mongo/bson/bsonobjbuilder.h
index 1cb8e7fbb51..3713cf8c313 100644
--- a/src/mongo/bson/bsonobjbuilder.h
+++ b/src/mongo/bson/bsonobjbuilder.h
@@ -44,6 +44,7 @@
 #include "mongo/bson/bsonelement.h"
 #include "mongo/bson/bsonmisc.h"
 #include "mongo/bson/bsonobj.h"
+#include "mongo/platform/decimal128.h"
 
 namespace mongo {
 
@@ -235,6 +236,14 @@ public:
         return append(fieldName, (int)n);
     }
 
+    /** Append a NumberDecimal */
+    BSONObjBuilder& append(StringData fieldName, Decimal128 n) {
+        _b.appendNum(static_cast<char>(NumberDecimal));
+        _b.appendStr(fieldName);
+        _b.appendNum(n);
+        return *this;
+    }
+
     /** Append a NumberLong */
     BSONObjBuilder& append(StringData fieldName, long long n) {
         _b.appendNum((char)NumberLong);
@@ -270,7 +279,6 @@ public:
 
     BSONObjBuilder& appendNumber(StringData fieldName, size_t n) {
         static const size_t maxInt = (1 << 30);
-
         if (n < maxInt)
             append(fieldName, static_cast<int>(n));
         else
@@ -278,6 +286,10 @@ public:
         return *this;
     }
 
+    BSONObjBuilder& appendNumber(StringData fieldName, Decimal128 decNumber) {
+        return append(fieldName, decNumber);
+    }
+
     BSONObjBuilder& appendNumber(StringData fieldName, long long llNumber) {
         static const long long maxInt = (1LL << 30);
         static const long long minInt = -maxInt;
@@ -895,7 +907,6 @@ inline BSONObjBuilder& BSONObjBuilder::append(StringData fieldName, const std::m
     return *this;
 }
 
-
 template <class L>
 inline BSONArrayBuilder& _appendArrayIt(BSONArrayBuilder& _this, const L& vals) {
     for (typename L::const_iterator i = vals.begin(); i != vals.end(); i++)
@@ -920,7 +931,6 @@ inline BSONFieldValue<BSONObj> BSONField<T>::query(const char* q, const T& t) co
     return BSONFieldValue<BSONObj>(_name, b.obj());
 }
 
-
 // $or helper: OR(BSON("x" << GT << 7), BSON("y" << LT 6));
 inline BSONObj OR(const BSONObj& a, const BSONObj& b) {
     return BSON("$or" << BSON_ARRAY(a << b));
@@ -962,7 +972,6 @@ inline BSONObjBuilder& BSONObjBuilderValueStream::operator<<(const UndefinedLabe
     return *_builder;
 }
 
-
 inline BSONObjBuilder& BSONObjBuilderValueStream::operator<<(const MinKeyLabeler& id) {
     _builder->appendMinKey(_fieldName);
     _fieldName = StringData();
@@ -1001,4 +1010,5 @@ inline BSONObjBuilder& BSONObjBuilder::appendTimestamp(StringData fieldName,
                                                        unsigned long long val) {
     return append(fieldName, Timestamp(val));
 }
-}
+
+}  // namespace mongo
diff --git a/src/mongo/bson/bsontypes.cpp b/src/mongo/bson/bsontypes.cpp
index bfea81ad58c..1e5895c3130 100644
--- a/src/mongo/bson/bsontypes.cpp
+++ b/src/mongo/bson/bsontypes.cpp
@@ -82,6 +82,8 @@ const char* typeName(BSONType type) {
             return "Timestamp";
         case NumberLong:
             return "NumberLong64";
+        case NumberDecimal:
+            return "NumberDecimal128";
         // JSTypeMax doesn't make sense to turn into a string; overlaps with highest-valued type
         case MaxKey:
             return "MaxKey";
@@ -89,4 +91,5 @@ const char* typeName(BSONType type) {
             return "Invalid";
     }
 }
+
 }  // namespace mongo
diff --git a/src/mongo/bson/bsontypes.h b/src/mongo/bson/bsontypes.h
index 9bd70822c8a..62fcf5b8ac4 100644
--- a/src/mongo/bson/bsontypes.h
+++ b/src/mongo/bson/bsontypes.h
@@ -29,7 +29,9 @@
 
 #pragma once
 
+#include "mongo/platform/decimal128.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/config.h"
 
 namespace mongo {
 
@@ -91,8 +93,10 @@ enum BSONType {
     bsonTimestamp = 17,
     /** 64 bit integer */
     NumberLong = 18,
+    /** 128 bit decimal */
+    NumberDecimal = 19,
     /** max type that is not MaxKey */
-    JSTypeMax = 18,
+    JSTypeMax = Decimal128::enabled ? 19 : 18,
     /** larger than all other types */
     MaxKey = 127
 };
@@ -131,6 +135,7 @@ inline int canonicalizeBSONType(BSONType type) {
             return 0;
         case jstNULL:
             return 5;
+        case NumberDecimal:
         case NumberDouble:
         case NumberInt:
         case NumberLong:
diff --git a/src/mongo/bson/json.cpp b/src/mongo/bson/json.cpp
index 55297486ef3..600a6dbff72 100644
--- a/src/mongo/bson/json.cpp
+++ b/src/mongo/bson/json.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/base/parse_number.h"
 #include "mongo/db/jsobj.h"
+#include "mongo/platform/decimal128.h"
 #include "mongo/platform/strtoll.h"
 #include "mongo/util/base64.h"
 #include "mongo/util/hex.h"
@@ -71,6 +72,7 @@ enum {
     NS_RESERVE_SIZE = 64,
     DB_RESERVE_SIZE = 64,
     NUMBERLONG_RESERVE_SIZE = 64,
+    NUMBERDECIMAL_RESERVE_SIZE = 64,
     DATE_RESERVE_SIZE = 64
 };
 
@@ -133,6 +135,11 @@ Status JParse::value(StringData fieldName, BSONObjBuilder& builder) {
         if (ret != Status::OK()) {
             return ret;
         }
+    } else if (readToken("NumberDecimal")) {
+        Status ret = numberDecimal(fieldName, builder);
+        if (ret != Status::OK()) {
+            return ret;
+        }
     } else if (readToken("Dbref") || readToken("DBRef")) {
         Status ret = dbRef(fieldName, builder);
         if (ret != Status::OK()) {
@@ -265,6 +272,14 @@ Status JParse::object(StringData fieldName, BSONObjBuilder& builder, bool subObj
         if (ret != Status::OK()) {
             return ret;
         }
+    } else if (firstField == "$numberDecimal") {
+        if (!subObject) {
+            return parseError("Reserved field name in base object: $numberDecimal");
+        }
+        Status ret = numberDecimalObject(fieldName, builder);
+        if (ret != Status::OK()) {
+            return ret;
+        }
     } else if (firstField == "$minKey") {
         if (!subObject) {
             return parseError("Reserved field name in base object: $minKey");
@@ -470,6 +485,7 @@ Status JParse::timestampObject(StringData fieldName, BSONObjBuilder& builder) {
     if (!readField("t")) {
         return parseError("Expected field name \"t\" in \"$timestamp\" sub object");
     }
+
     if (!readToken(COLON)) {
         return parseError("Expecting ':'");
     }
@@ -638,6 +654,25 @@ Status JParse::numberLongObject(StringData fieldName, BSONObjBuilder& builder) {
     return Status::OK();
 }
 
+Status JParse::numberDecimalObject(StringData fieldName, BSONObjBuilder& builder) {
+    if (!readToken(COLON)) {
+        return parseError("Expecting ':'");
+    }
+    // The number must be a quoted string, since large decimal numbers could overflow other types
+    // and thus may not be valid JSON
+    std::string numberDecimalString;
+    numberDecimalString.reserve(NUMBERDECIMAL_RESERVE_SIZE);
+    Status ret = quotedString(&numberDecimalString);
+    if (!ret.isOK()) {
+        return ret;
+    }
+
+    Decimal128 numberDecimal(numberDecimalString);
+
+    builder.appendNumber(fieldName, numberDecimal);
+    return Status::OK();
+}
+
 Status JParse::minKeyObject(StringData fieldName, BSONObjBuilder& builder) {
     if (!readToken(COLON)) {
         return parseError("Expecting ':'");
@@ -823,6 +858,26 @@ Status JParse::numberLong(StringData fieldName, BSONObjBuilder& builder) {
     return Status::OK();
 }
 
+Status JParse::numberDecimal(StringData fieldName, BSONObjBuilder& builder) {
+    if (!readToken(LPAREN)) {
+        return parseError("Expecting '('");
+    }
+
+    std::string decString;
+    decString.reserve(NUMBERDECIMAL_RESERVE_SIZE);
+    Status ret = quotedString(&decString);
+    if (ret != Status::OK()) {
+        return ret;
+    }
+    Decimal128 val(decString);
+
+    if (!readToken(RPAREN)) {
+        return parseError("Expecting ')'");
+    }
+    builder.appendNumber(fieldName, val);
+    return Status::OK();
+}
+
 Status JParse::numberInt(StringData fieldName, BSONObjBuilder& builder) {
     if (!readToken(LPAREN)) {
         return parseError("Expecting '('");
@@ -846,7 +901,6 @@ Status JParse::numberInt(StringData fieldName, BSONObjBuilder& builder) {
     return Status::OK();
 }
 
-
 Status JParse::dbRef(StringData fieldName, BSONObjBuilder& builder) {
     BSONObjBuilder subBuilder(builder.subobjStart(fieldName));
 
diff --git a/src/mongo/bson/json.h b/src/mongo/bson/json.h
index 7a22d1d2186..8553f79408e 100644
--- a/src/mongo/bson/json.h
+++ b/src/mongo/bson/json.h
@@ -107,6 +107,7 @@ public:
      *   | NUMBER
      *   | NUMBERINT
      *   | NUMBERLONG
+     *   | NUMBERDECIMAL
      *   | OBJECT
      *   | ARRAY
      *
@@ -152,6 +153,7 @@ private:
      *   | REFOBJECT
      *   | UNDEFINEDOBJECT
      *   | NUMBERLONGOBJECT
+     *   | NUMBERDECIMALOBJECT
      *   | MINKEYOBJECT
      *   | MAXKEYOBJECT
      *
@@ -225,6 +227,12 @@ private:
     Status numberLongObject(StringData fieldName, BSONObjBuilder&);
 
     /*
+     * NUMBERDECIMALOBJECT :
+     *     { FIELD("$numberDecimal") : "<number>" }
+     */
+    Status numberDecimalObject(StringData fieldName, BSONObjBuilder&);
+
+    /*
      * MINKEYOBJECT :
      *     { FIELD("$minKey") : 1 }
      */
@@ -282,6 +290,12 @@ private:
     Status numberLong(StringData fieldName, BSONObjBuilder&);
 
     /*
+     * NUMBERDECIMAL :
+     *     NumberDecimal( <number> )
+     */
+    Status numberDecimal(StringData fieldName, BSONObjBuilder&);
+
+    /*
      * NUMBERINT :
      *     NumberInt( <number> )
      */
diff --git a/src/mongo/bson/util/builder.h b/src/mongo/bson/util/builder.h
index 89c726b38d4..f4248efbe84 100644
--- a/src/mongo/bson/util/builder.h
+++ b/src/mongo/bson/util/builder.h
@@ -40,6 +40,7 @@
 #include "mongo/base/data_view.h"
 #include "mongo/base/string_data.h"
 #include "mongo/bson/inline_decls.h"
+#include "mongo/platform/decimal128.h"
 #include "mongo/util/allocator.h"
 #include "mongo/util/assert_util.h"
 
@@ -52,8 +53,6 @@ namespace mongo {
 struct PackedDouble {
     double d;
 } PACKED_DECL;
-
-
 /* Note the limit here is rather arbitrary and is simply a standard. generally the code works
    with any object that fits in ram.
 
@@ -220,6 +219,10 @@ public:
     void appendNum(unsigned long long j) {
         appendNumImpl(j);
     }
+    void appendNum(Decimal128 j) {
+        BOOST_STATIC_ASSERT(sizeof(Decimal128::Value) == 16);
+        appendNumImpl(j.getValue());
+    }
 
     void appendBuf(const void* src, size_t len) {
         memcpy(grow((int)len), src, len);
@@ -290,8 +293,6 @@ private:
         // we bake that assumption in here. This decision should be revisited soon.
         DataView(grow(sizeof(t))).write(tagLittleEndian(t));
     }
-
-
     /* "slow" portion of 'grow()'  */
     void NOINLINE_DECL grow_reallocate(int minSize) {
         int a = 64;
diff --git a/src/mongo/db/matcher/expression_leaf.cpp b/src/mongo/db/matcher/expression_leaf.cpp
index 3b53ecc5ba6..79836a43f2c 100644
--- a/src/mongo/db/matcher/expression_leaf.cpp
+++ b/src/mongo/db/matcher/expression_leaf.cpp
@@ -36,6 +36,7 @@
 
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/bsonmisc.h"
+#include "mongo/config.h"
 #include "mongo/db/field_ref.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/matcher/path.h"
@@ -395,6 +396,9 @@ const std::string TypeMatchExpression::kMatchesAllNumbersAlias = "number";
 
 const std::unordered_map<std::string, BSONType> TypeMatchExpression::typeAliasMap = {
     {"double", NumberDouble},
+#ifdef MONGO_CONFIG_EXPERIMENTAL_DECIMAL_SUPPORT
+    {"decimal", NumberDecimal},
+#endif
     {"string", String},
     {"object", Object},
     {"array", Array},
diff --git a/src/mongo/db/pipeline/value.cpp b/src/mongo/db/pipeline/value.cpp
index 3f5a9377dd4..6c2dc767256 100644
--- a/src/mongo/db/pipeline/value.cpp
+++ b/src/mongo/db/pipeline/value.cpp
@@ -37,6 +37,7 @@
 #include "mongo/base/compare_numbers.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/pipeline/document.h"
+#include "mongo/platform/decimal128.h"
 #include "mongo/util/hex.h"
 #include "mongo/util/mongoutils/str.h"
 
@@ -76,6 +77,7 @@ void ValueStorage::verifyRefCountingIfShould() const {
             verify(refCounter == !shortStr);
             break;
 
+        case NumberDecimal:
         case BinData:  // TODO this should probably support short-string optimization
         case Array:    // TODO this should probably support empty-is-NULL optimization
         case DBRef:
@@ -206,6 +208,10 @@ Value::Value(const BSONElement& elem) : _storage(elem.type()) {
             _storage.longValue = elem.numberLong();
             break;
 
+        case NumberDecimal:
+            _storage.putDecimal(elem.numberDecimal());
+            break;
+
         case CodeWScope: {
             StringData code(elem.codeWScopeCode(), elem.codeWScopeCodeLen() - 1);
             _storage.putCodeWScope(BSONCodeWScope(code, elem.codeWScopeObject()));
@@ -244,12 +250,26 @@ Value Value::createIntOrLong(long long longValue) {
     return Value(intValue);
 }
 
+Decimal128 Value::getDecimal() const {
+    BSONType type = getType();
+    if (type == NumberInt)
+        return Decimal128(_storage.intValue);
+    if (type == NumberLong)
+        return Decimal128(_storage.longValue);
+    if (type == NumberDouble)
+        return Decimal128(_storage.doubleValue);
+    invariant(type == NumberDecimal);
+    return _storage.getDecimal();
+}
+
 double Value::getDouble() const {
     BSONType type = getType();
     if (type == NumberInt)
         return _storage.intValue;
     if (type == NumberLong)
         return static_cast<double>(_storage.longValue);
+    if (type == NumberDecimal)
+        return _storage.getDecimal().toDouble();
 
     verify(type == NumberDouble);
     return _storage.doubleValue;
@@ -294,6 +314,8 @@ BSONObjBuilder& operator<<(BSONObjBuilderValueStream& builder, const Value& val)
             return builder << val.getLong();
         case NumberDouble:
             return builder << val.getDouble();
+        case NumberDecimal:
+            return builder << val.getDecimal();
         case String:
             return builder << val.getStringData();
         case Bool:
@@ -379,6 +401,8 @@ bool Value::coerceToBool() const {
             return _storage.longValue;
         case NumberDouble:
             return _storage.doubleValue;
+        case NumberDecimal:
+            return !_storage.getDecimal().isZero();
     }
     verify(false);
 }
@@ -394,6 +418,9 @@ int Value::coerceToInt() const {
         case NumberDouble:
             return static_cast<int>(_storage.doubleValue);
 
+        case NumberDecimal:
+            return (_storage.getDecimal()).toInt();
+
         default:
             uassert(16003,
                     str::stream() << "can't convert from BSON type " << typeName(getType())
@@ -413,6 +440,9 @@ long long Value::coerceToLong() const {
         case NumberDouble:
             return static_cast<long long>(_storage.doubleValue);
 
+        case NumberDecimal:
+            return (_storage.getDecimal()).toLong();
+
         default:
             uassert(16004,
                     str::stream() << "can't convert from BSON type " << typeName(getType())
@@ -432,6 +462,9 @@ double Value::coerceToDouble() const {
         case NumberLong:
             return static_cast<double>(_storage.longValue);
 
+        case NumberDecimal:
+            return (_storage.getDecimal()).toDouble();
+
         default:
             uassert(16005,
                     str::stream() << "can't convert from BSON type " << typeName(getType())
@@ -440,6 +473,28 @@ double Value::coerceToDouble() const {
     }  // switch(getType())
 }
 
+Decimal128 Value::coerceToDecimal() const {
+    switch (getType()) {
+        case NumberDecimal:
+            return _storage.getDecimal();
+
+        case NumberInt:
+            return Decimal128(_storage.intValue);
+
+        case NumberLong:
+            return Decimal128(_storage.longValue);
+
+        case NumberDouble:
+            return Decimal128(_storage.doubleValue);
+
+        default:
+            uassert(16008,
+                    str::stream() << "can't convert from BSON type " << typeName(getType())
+                                  << " to decimal",
+                    false);
+    }  // switch(getType())
+}
+
 long long Value::coerceToDate() const {
     switch (getType()) {
         case Date:
@@ -518,6 +573,9 @@ string Value::coerceToString() const {
         case NumberLong:
             return str::stream() << _storage.longValue;
 
+        case NumberDecimal:
+            return str::stream() << _storage.getDecimal().toString();
+
         case Code:
         case Symbol:
         case String:
@@ -602,6 +660,23 @@ int Value::compare(const Value& rL, const Value& rR) {
             return cmp(rL._storage.dateValue, rR._storage.dateValue);
 
         // Numbers should compare by equivalence even if different types
+
+        case NumberDecimal: {
+            switch (rType) {
+                case NumberDecimal:
+                    return compareDecimals(rL._storage.getDecimal(), rR._storage.getDecimal());
+                case NumberInt:
+                    return compareDecimalToInt(rL._storage.getDecimal(), rR._storage.intValue);
+                case NumberLong:
+                    return compareDecimalToLong(rL._storage.getDecimal(), rR._storage.longValue);
+                case NumberDouble:
+                    return compareDecimalToDouble(rL._storage.getDecimal(),
+                                                  rR._storage.doubleValue);
+                default:
+                    invariant(false);
+            }
+        }
+
         case NumberInt: {
             // All types can precisely represent all NumberInts, so it is safe to simply convert to
             // whatever rhs's type is.
@@ -612,6 +687,8 @@ int Value::compare(const Value& rL, const Value& rR) {
                     return compareLongs(rL._storage.intValue, rR._storage.longValue);
                 case NumberDouble:
                     return compareDoubles(rL._storage.intValue, rR._storage.doubleValue);
+                case NumberDecimal:
+                    return compareIntToDecimal(rL._storage.intValue, rR._storage.getDecimal());
                 default:
                     invariant(false);
             }
@@ -625,6 +702,8 @@ int Value::compare(const Value& rL, const Value& rR) {
                     return compareLongs(rL._storage.longValue, rR._storage.intValue);
                 case NumberDouble:
                     return compareLongToDouble(rL._storage.longValue, rR._storage.doubleValue);
+                case NumberDecimal:
+                    return compareLongToDecimal(rL._storage.longValue, rR._storage.getDecimal());
                 default:
                     invariant(false);
             }
@@ -638,6 +717,9 @@ int Value::compare(const Value& rL, const Value& rR) {
                     return compareDoubles(rL._storage.doubleValue, rR._storage.intValue);
                 case NumberLong:
                     return compareDoubleToLong(rL._storage.doubleValue, rR._storage.longValue);
+                case NumberDecimal:
+                    return compareDoubleToDecimal(rL._storage.doubleValue,
+                                                  rR._storage.getDecimal());
                 default:
                     invariant(false);
             }
@@ -737,9 +819,27 @@ void Value::hash_combine(size_t& seed) const {
             boost::hash_combine(seed, _storage.dateValue);
             break;
 
+        case mongo::NumberDecimal: {
+            const Decimal128 dcml = getDecimal();
+            if (dcml.toAbs().isGreater(Decimal128(std::numeric_limits<double>::max())) &&
+                !dcml.isInfinite() && !dcml.isNaN()) {
+                // Normalize our decimal to force equivalent decimals
+                // in the same cohort to hash to the same value
+                Decimal128 dcmlNorm(dcml.normalize());
+                boost::hash_combine(seed, dcmlNorm.getValue().low64);
+                boost::hash_combine(seed, dcmlNorm.getValue().high64);
+                break;
+            }
+            // Else, fall through and convert the decimal to a double and hash.
+            // At this point the decimal fits into the range of doubles, is infinity, or is NaN,
+            // which doubles have a cheaper representation for.
+        }
         // This converts all numbers to doubles, which ignores the low-order bits of
-        // NumberLongs > 2**53, but that is ok since the hash will still be the same for
-        // equal numbers and is still likely to be different for different numbers.
+        // NumberLongs > 2**53 and precise decimal numbers without double representations,
+        // but that is ok since the hash will still be the same for equal numbers and is
+        // still likely to be different for different numbers. (Note: this issue only
+        // applies for decimals when they are inside of the valid double range. See
+        // the above case.)
         // SERVER-16851
         case NumberDouble:
         case NumberLong:
@@ -807,6 +907,9 @@ void Value::hash_combine(size_t& seed) const {
 BSONType Value::getWidestNumeric(BSONType lType, BSONType rType) {
     if (lType == NumberDouble) {
         switch (rType) {
+            case NumberDecimal:
+                return NumberDecimal;
+
             case NumberDouble:
             case NumberLong:
             case NumberInt:
@@ -817,6 +920,9 @@ BSONType Value::getWidestNumeric(BSONType lType, BSONType rType) {
         }
     } else if (lType == NumberLong) {
         switch (rType) {
+            case NumberDecimal:
+                return NumberDecimal;
+
             case NumberDouble:
                 return NumberDouble;
 
@@ -829,6 +935,9 @@ BSONType Value::getWidestNumeric(BSONType lType, BSONType rType) {
         }
     } else if (lType == NumberInt) {
         switch (rType) {
+            case NumberDecimal:
+                return NumberDecimal;
+
             case NumberDouble:
                 return NumberDouble;
 
@@ -841,12 +950,25 @@ BSONType Value::getWidestNumeric(BSONType lType, BSONType rType) {
             default:
                 break;
         }
+    } else if (lType == NumberDecimal) {
+        switch (rType) {
+            case NumberInt:
+            case NumberLong:
+            case NumberDouble:
+            case NumberDecimal:
+                return NumberDecimal;
+
+            default:
+                break;
+        }
     }
 
     // Reachable, but callers must subsequently err out in this case.
     return Undefined;
 }
 
+// TODO: Add Decimal128 support to Value::integral()
+// SERVER-19735
 bool Value::integral() const {
     switch (getType()) {
         case NumberInt:
@@ -894,6 +1016,9 @@ size_t Value::getApproximateSize() const {
         case DBRef:
             return sizeof(Value) + sizeof(RCDBRef) + _storage.getDBRef()->ns.size();
 
+        case NumberDecimal:
+            return sizeof(Value) + sizeof(RCDecimal);
+
         // These types are always contained within the Value
         case EOO:
         case MinKey:
@@ -939,6 +1064,8 @@ ostream& operator<<(ostream& out, const Value& val) {
             return out << "Code(\"" << val.getCode() << "\")";
         case Bool:
             return out << (val.getBool() ? "true" : "false");
+        case NumberDecimal:
+            return out << val.getDecimal().toString();
         case NumberDouble:
             return out << val.getDouble();
         case NumberLong:
@@ -1009,6 +1136,9 @@ void Value::serializeForSorter(BufBuilder& buf) const {
         case NumberDouble:
             buf.appendNum(_storage.doubleValue);
             break;
+        case NumberDecimal:
+            buf.appendNum(_storage.getDecimal());
+            break;
         case Bool:
             buf.appendChar(_storage.boolValue);
             break;
@@ -1090,6 +1220,8 @@ Value Value::deserializeForSorter(BufReader& buf, const SorterDeserializeSetting
             return Value(buf.read<long long>());
         case NumberDouble:
             return Value(buf.read<double>());
+        case NumberDecimal:
+            return Value(buf.read<Decimal128>());
         case Bool:
             return Value(bool(buf.read<char>()));
         case Date:
diff --git a/src/mongo/db/pipeline/value.h b/src/mongo/db/pipeline/value.h
index 6c27c04078d..832b45d408e 100644
--- a/src/mongo/db/pipeline/value.h
+++ b/src/mongo/db/pipeline/value.h
@@ -71,6 +71,7 @@ public:
     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 Decimal128& value) : _storage(NumberDecimal, value) {}
     explicit Value(const Timestamp& value) : _storage(bsonTimestamp, value) {}
     explicit Value(const OID& value) : _storage(jstOID, value) {}
     explicit Value(StringData value) : _storage(String, value) {}
@@ -118,6 +119,8 @@ public:
     }
 
     /// true if type represents a number
+    // TODO: Add _storage.type == NumberDecimal
+    // SERVER-19735
     bool numeric() const {
         return _storage.type == NumberDouble || _storage.type == NumberLong ||
             _storage.type == NumberInt;
@@ -138,6 +141,7 @@ public:
      *  Asserts if the requested value type is not exactly correct.
      *  See coerceTo methods below for a more type-flexible alternative.
      */
+    Decimal128 getDecimal() const;
     double getDouble() const;
     std::string getString() const;
     Document getDocument() const;
@@ -184,6 +188,7 @@ public:
     int coerceToInt() const;
     long long coerceToLong() const;
     double coerceToDouble() const;
+    Decimal128 coerceToDecimal() const;
     Timestamp coerceToTimestamp() const;
     long long coerceToDate() const;
     time_t coerceToTimeT() const;
diff --git a/src/mongo/db/pipeline/value_internal.h b/src/mongo/db/pipeline/value_internal.h
index 142b94368b3..21201a50c6e 100644
--- a/src/mongo/db/pipeline/value_internal.h
+++ b/src/mongo/db/pipeline/value_internal.h
@@ -68,6 +68,12 @@ public:
     const OID oid;
 };
 
+class RCDecimal : public RefCountable {
+public:
+    RCDecimal(const Decimal128& decVal) : decimalValue(decVal) {}
+    const Decimal128 decimalValue;
+};
+
 #pragma pack(1)
 class ValueStorage {
 public:
@@ -99,6 +105,11 @@ public:
         type = t;
         doubleValue = d;
     }
+    ValueStorage(BSONType t, const Decimal128& d) {
+        zero();
+        type = t;
+        putDecimal(d);
+    }
     ValueStorage(BSONType t, Timestamp r) {
         zero();
         type = t;
@@ -201,6 +212,10 @@ public:
         putRefCountable(new RCCodeWScope(cws.code.toString(), cws.scope));
     }
 
+    void putDecimal(const Decimal128& d) {
+        putRefCountable(new RCDecimal(d));
+    }
+
     void putRefCountable(boost::intrusive_ptr<const RefCountable> ptr) {
         genericRCPtr = ptr.get();
 
@@ -237,6 +252,12 @@ public:
         return static_cast<const RCDBRef*>(genericRCPtr);
     }
 
+    Decimal128 getDecimal() const {
+        dassert(typeid(*genericRCPtr) == typeid(const RCDecimal));
+        const RCDecimal* decPtr = static_cast<const RCDecimal*>(genericRCPtr);
+        return decPtr->decimalValue;
+    }
+
     // Document is incomplete here so this can't be inline
     Document getDocument() const;
 
diff --git a/src/mongo/db/storage/key_string_test.cpp b/src/mongo/db/storage/key_string_test.cpp
index 082d01366a9..7cccd712c67 100644
--- a/src/mongo/db/storage/key_string_test.cpp
+++ b/src/mongo/db/storage/key_string_test.cpp
@@ -356,6 +356,10 @@ TEST(KeyStringTest, Timestamp) {
 
 TEST(KeyStringTest, AllTypesRoundtrip) {
     for (int i = 1; i <= JSTypeMax; i++) {
+        // TODO: Currently KeyString does not support NumberDecimal
+        // SERVER-19703
+        if (i == NumberDecimal)
+            continue;
         {
             BSONObjBuilder b;
             b.appendMinForType("", i);
diff --git a/src/mongo/platform/decimal128.h b/src/mongo/platform/decimal128.h
index c186265dc2a..7a1188a8904 100644
--- a/src/mongo/platform/decimal128.h
+++ b/src/mongo/platform/decimal128.h
@@ -51,9 +51,9 @@ public:
  * and is set by the build flag --experimental-decimal-support.
  */
 #ifdef MONGO_CONFIG_EXPERIMENTAL_DECIMAL_SUPPORT
-    const bool enabled = true;
+    static const bool enabled = true;
 #else
-    const bool enabled = false;
+    static const bool enabled = false;
 #endif
 
     /**