SERVER-53982 RecordId supports ObjectId type

Allow RecordId to store either an int64_t or ObjectId. This type is
compact and cheap to copy. It assumes that callers who attempt to
observe the underlying data are aware of its underlying type.

1 files changed, 194 insertions, 62 deletions

diff --git a/src/mongo/db/record_id.h b/src/mongo/db/record_id.h
index ac816a44291..78953f29d6d 100644
--- a/src/mongo/db/record_id.h
+++ b/src/mongo/db/record_id.h
@@ -37,69 +37,148 @@
 #include <ostream>
 
 #include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/bson/oid.h"
 #include "mongo/bson/util/builder.h"
 #include "mongo/util/bufreader.h"
 
 namespace mongo {
 
+
 /**
  * The key that uniquely identifies a Record in a Collection or RecordStore.
  */
 class RecordId {
 public:
-    // This set of constants define the boundaries of the 'normal' and 'reserved' id ranges.
-    static constexpr int64_t kNullRepr = 0;
+    // This set of constants define the boundaries of the 'normal' and 'reserved' id ranges for
+    // the kLong format.
     static constexpr int64_t kMinRepr = LLONG_MIN;
     static constexpr int64_t kMaxRepr = LLONG_MAX;
     static constexpr int64_t kMinReservedRepr = kMaxRepr - (1024 * 1024);
 
+    // OID Constants
+    static constexpr unsigned char kMinOID[OID::kOIDSize] = {0x00};
+    static constexpr unsigned char kMaxOID[OID::kOIDSize] = {
+        0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff};
+    // This reserved range leaves 2^20 possible reserved values.
+    static constexpr unsigned char kMinReservedOID[OID::kOIDSize] = {
+        0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xf0, 0x00, 0x00};
+
     /**
-     * Enumerates all ids in the reserved range that have been allocated for a specific purpose.
+     * A RecordId that compares less than all ids for a given data format.
      */
-    enum class ReservedId : int64_t { kWildcardMultikeyMetadataId = kMinReservedRepr };
+    template <typename T>
+    static RecordId min() {
+        if constexpr (std::is_same_v<T, int64_t>) {
+            return RecordId(kMinRepr);
+        } else {
+            static_assert(std::is_same_v<T, OID>, "Unsupported RecordID format");
+            return RecordId(OID(kMinOID));
+        }
+    }
 
     /**
-     * Constructs a Null RecordId.
+     * A RecordId that compares greater than all ids that represent documents in a collection.
      */
-    RecordId() : _repr(kNullRepr) {}
-
-    explicit RecordId(int64_t repr) : _repr(repr) {}
+    template <typename T>
+    static RecordId max() {
+        if constexpr (std::is_same_v<T, int64_t>) {
+            return RecordId(kMaxRepr);
+        } else {
+            static_assert(std::is_same_v<T, OID>, "Unsupported RecordID format");
+            return RecordId(OID(kMaxOID));
+        }
+    }
 
-    explicit RecordId(ReservedId repr) : RecordId(static_cast<int64_t>(repr)) {}
+    /**
+     * Returns the first record in the reserved id range at the top of the RecordId space.
+     */
+    template <typename T>
+    static RecordId minReserved() {
+        if constexpr (std::is_same_v<T, int64_t>) {
+            return RecordId(kMinReservedRepr);
+        } else {
+            static_assert(std::is_same_v<T, OID>, "Unsupported RecordID format");
+            return RecordId(OID(kMinReservedOID));
+        }
+    }
 
     /**
-     * Construct a RecordId from two halves.
-     * TODO consider removing.
+     * Enumerates all reserved ids that have been allocated for a specific purpose.
+     * The underlying value of the reserved Record ID is data-format specific and must be retrieved
+     * by the getReservedId() helper.
      */
-    RecordId(int high, int low) : _repr((uint64_t(high) << 32) | uint32_t(low)) {}
+    enum class Reservation { kWildcardMultikeyMetadataId };
 
     /**
-     * A RecordId that compares less than all ids that represent documents in a collection.
+     * Returns the reserved RecordId value for a given Reservation.
      */
-    static RecordId min() {
-        return RecordId(kMinRepr);
+    template <typename T>
+    static RecordId reservedIdFor(Reservation res) {
+        // There is only one reservation at the moment.
+        invariant(res == Reservation::kWildcardMultikeyMetadataId);
+        if constexpr (std::is_same_v<T, int64_t>) {
+            return RecordId(kMinReservedRepr);
+        } else {
+            static_assert(std::is_same_v<T, OID>, "Unsupported RecordID format");
+            return RecordId(OID(kMinReservedOID));
+        }
     }
 
+    RecordId() : _format(Format::kNull) {}
+    explicit RecordId(int64_t repr) : _storage(repr), _format(Format::kLong) {}
+    explicit RecordId(const OID& oid) : _storage(oid), _format(Format::kOid) {}
+
     /**
-     * A RecordId that compares greater than all ids that represent documents in a collection.
+     * Construct a RecordId from two halves.
      */
-    static RecordId max() {
-        return RecordId(kMaxRepr);
-    }
+    RecordId(int high, int low) : RecordId((uint64_t(high) << 32) | uint32_t(low)) {}
+
+    /** Tag for dispatching on null values */
+    class Null {};
 
     /**
-     * Returns the first record in the reserved id range at the top of the RecordId space.
+     * Helpers to dispatch based on the underlying type.
      */
-    static RecordId minReserved() {
-        return RecordId(kMinReservedRepr);
+    template <typename OnNull, typename OnLong, typename OnOid>
+    auto withFormat(OnNull&& onNull, OnLong&& onLong, OnOid&& onOid) const {
+        switch (_format) {
+            case Format::kNull:
+                return onNull(Null());
+            case Format::kLong:
+                return onLong(_storage._long);
+            case Format::kOid:
+                return onOid(_storage._oid);
+            default:
+                MONGO_UNREACHABLE;
+        }
     }
 
-    bool isNull() const {
-        return _repr == 0;
+    /**
+     * Returns the underlying data for a given format. Will invariant if the RecordId is not storing
+     * requested format.
+     */
+    template <typename T>
+    T as() const {
+        if constexpr (std::is_same_v<T, int64_t>) {
+            // In the the int64_t format, null can also be represented by '0'.
+            if (_format == Format::kNull) {
+                return 0;
+            }
+            invariant(_format == Format::kLong);
+            return _storage._long;
+        } else {
+            static_assert(std::is_same_v<T, OID>, "Unsupported RecordID format");
+            invariant(_format == Format::kOid);
+            return _storage._oid;
+        }
     }
 
-    int64_t repr() const {
-        return _repr;
+    bool isNull() const {
+        // In the the int64_t format, null can also represented by '0'.
+        if (_format == Format::kLong) {
+            return _storage._long == 0;
+        }
+        return _format == Format::kNull;
     }
 
     /**
@@ -116,18 +195,55 @@ public:
      * excluding the reserved range at the top of the RecordId space.
      */
     bool isNormal() const {
-        return _repr > 0 && _repr < kMinReservedRepr;
+        return withFormat([](Null n) { return false; },
+                          [](int64_t rid) { return rid > 0 && rid < kMinReservedRepr; },
+                          [](const OID& oid) { return oid.compare(OID(kMinReservedOID)) < 0; });
     }
 
     /**
      * Returns true if this RecordId falls within the reserved range at the top of the record space.
      */
     bool isReserved() const {
-        return _repr >= kMinReservedRepr && _repr < kMaxRepr;
+        return withFormat([](Null n) { return false; },
+                          [](int64_t rid) { return rid >= kMinReservedRepr && rid < kMaxRepr; },
+                          [](const OID& oid) {
+                              return oid.compare(OID(kMinReservedOID)) >= 0 &&
+                                  oid.compare(OID(kMaxOID)) < 0;
+                          });
+    }
+
+    int compare(const RecordId& rhs) const {
+        // Null always compares less than every other RecordId.
+        if (isNull() && rhs.isNull()) {
+            return 0;
+        } else if (isNull()) {
+            return -1;
+        } else if (rhs.isNull()) {
+            return 1;
+        }
+        invariant(_format == rhs._format);
+        return withFormat([](Null n) { return 0; },
+                          [&](const int64_t rid) {
+                              return rid == rhs._storage._long ? 0
+                                                               : rid < rhs._storage._long ? -1 : 1;
+                          },
+                          [&](const OID& oid) { return oid.compare(rhs._storage._oid); });
+    }
+
+    size_t hash() const {
+        size_t hash = 0;
+        withFormat([](Null n) {},
+                   [&](int64_t rid) { boost::hash_combine(hash, rid); },
+                   [&](const OID& oid) {
+                       boost::hash_combine(hash, std::string(oid.view().view(), OID::kOIDSize));
+                   });
+        return hash;
     }
 
-    int compare(RecordId rhs) const {
-        return _repr == rhs._repr ? 0 : _repr < rhs._repr ? -1 : 1;
+    std::string toString() const {
+        return withFormat([](Null n) { return std::string("null"); },
+                          [](int64_t rid) { return std::to_string(rid); },
+                          [](const OID& oid) { return oid.toString(); });
     }
 
     /**
@@ -135,70 +251,86 @@ public:
      * may differ across platforms. Hash values should not be persisted.
      */
     struct Hasher {
-        size_t operator()(RecordId rid) const {
-            size_t hash = 0;
-            // TODO consider better hashes
-            boost::hash_combine(hash, rid.repr());
-            return hash;
+        size_t operator()(const RecordId& rid) const {
+            return rid.hash();
         }
     };
 
-    /// members for Sorter
-    struct SorterDeserializeSettings {};  // unused
-    void serializeForSorter(BufBuilder& buf) const {
-        buf.appendNum(static_cast<long long>(_repr));
-    }
-    static RecordId deserializeForSorter(BufReader& buf, const SorterDeserializeSettings&) {
-        return RecordId(buf.read<LittleEndian<int64_t>>());
-    }
-    int memUsageForSorter() const {
-        return sizeof(RecordId);
-    }
-    RecordId getOwned() const {
-        return *this;
-    }
-
     void serialize(fmt::memory_buffer& buffer) const {
-        fmt::format_to(buffer, "RecordId({})", _repr);
+        withFormat([&](Null n) { fmt::format_to(buffer, "RecordId(null)"); },
+                   [&](int64_t rid) { fmt::format_to(buffer, "RecordId({})", rid); },
+                   [&](const OID& oid) { fmt::format_to(buffer, "RecordId({})", oid.toString()); });
     }
 
     void serialize(BSONObjBuilder* builder) const {
-        builder->append("RecordId"_sd, _repr);
+        withFormat([&](Null n) { builder->append("RecordId", "null"); },
+                   [&](int64_t rid) { builder->append("RecordId"_sd, rid); },
+                   [&](const OID& oid) { builder->append("RecordId"_sd, oid); });
     }
 
 private:
-    int64_t _repr;
+    /**
+     * Specifies the storage format of this RecordId.
+     */
+    enum class Format : uint32_t {
+        /** Contains no value */
+        kNull,
+        /** int64_t */
+        kLong,
+        /** OID = char[12] */
+        kOid
+    };
+
+// Pack our union so that it only uses 12 bytes. The union will default to a 8 byte alignment,
+// making it 16 bytes total with 4 bytes of padding. Instead, we force the union to use a 4 byte
+// alignment, so it packs into 12 bytes. This leaves 4 bytes for our Format, allowing the RecordId
+// to use 16 bytes total.
+#pragma pack(push, 4)
+    union Storage {
+        // Format::kLong
+        int64_t _long;
+        // Format::kOid
+        OID _oid;
+
+        Storage() {}
+        Storage(int64_t s) : _long(s) {}
+        Storage(const OID& s) : _oid(s) {}
+    };
+#pragma pack(pop)
+
+    Storage _storage;
+    Format _format;
 };
 
 inline bool operator==(RecordId lhs, RecordId rhs) {
-    return lhs.repr() == rhs.repr();
+    return lhs.compare(rhs) == 0;
 }
 inline bool operator!=(RecordId lhs, RecordId rhs) {
-    return lhs.repr() != rhs.repr();
+    return lhs.compare(rhs);
 }
 inline bool operator<(RecordId lhs, RecordId rhs) {
-    return lhs.repr() < rhs.repr();
+    return lhs.compare(rhs) < 0;
 }
 inline bool operator<=(RecordId lhs, RecordId rhs) {
-    return lhs.repr() <= rhs.repr();
+    return lhs.compare(rhs) <= 0;
 }
 inline bool operator>(RecordId lhs, RecordId rhs) {
-    return lhs.repr() > rhs.repr();
+    return lhs.compare(rhs) > 0;
 }
 inline bool operator>=(RecordId lhs, RecordId rhs) {
-    return lhs.repr() >= rhs.repr();
+    return lhs.compare(rhs) >= 0;
 }
 
 inline StringBuilder& operator<<(StringBuilder& stream, const RecordId& id) {
-    return stream << "RecordId(" << id.repr() << ')';
+    return stream << "RecordId(" << id.toString() << ')';
 }
 
 inline std::ostream& operator<<(std::ostream& stream, const RecordId& id) {
-    return stream << "RecordId(" << id.repr() << ')';
+    return stream << "RecordId(" << id.toString() << ')';
 }
 
 inline std::ostream& operator<<(std::ostream& stream, const boost::optional<RecordId>& id) {
-    return stream << "RecordId(" << (id ? id.get().repr() : 0) << ')';
+    return stream << "RecordId(" << (id ? id->toString() : 0) << ')';
 }
 
 }  // namespace mongo