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

12 files changed, 455 insertions, 24 deletions

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;